From e1072f06b58b05158b4220ac7dd34916764ae47d Mon Sep 17 00:00:00 2001 From: nichinichisou Date: Sun, 18 Oct 2026 20:26:26 +0800 Subject: [PATCH 1/2] fix(depot): retain planned and standalone task results --- CHANGELOG.md | 1 + docs/assets.md | 9 +++++++- docs/contracts.md | 4 ++++ src/nnnotes/cli/assets.py | 29 ++++++++++++++++------- src/nnnotes/pipeline/orchestrate.py | 7 ++++-- src/nnnotes/pipeline/plan.py | 25 ++++++++++++++++---- src/nnnotes/schemas/layout.schema.json | 12 ++++++++++ tests/cli/test_assets.py | 11 ++++++++- tests/pipeline/test_orchestrate.py | 9 ++++++++ tests/pipeline/test_plan.py | 32 ++++++++++++++++++++++++-- 10 files changed, 120 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91c8ea9..126e90d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ schemas in `nnnotes.schemas` and the documented Python API. Before 1.0 a minor v ### Features - Share catalog versions and labels through depot remotes, with verified catalog files fetched on demand. +- Retain census planning results, emitted task descriptions and standalone task runs as depot collection roots. - Store content, task inputs, results and catalog versions in a persistent depot with optional remote storage, verified collection and bounded source downloads. - Release recoverable bundle downloads after successful conversion batches under the configured budget and command leases. - Assemble Live2D, room, spot, live, story and UI products from shared object tables and explicit manifest references. diff --git a/docs/assets.md b/docs/assets.md index e3fb2ff..f7743a1 100644 --- a/docs/assets.md +++ b/docs/assets.md @@ -91,7 +91,7 @@ gives per stage its counts and its first 20 tasks that are not hits; `--json` ha | `--json` | the plan document (`nnnotes.plan/1`) | | `--check` | exit 1 when anything would run or a layout file would change, else 0 | | `--census` | first fetch the bundles without a census and take it, so the plan knows every bundle's classes | -| `--emit-tasks DIR` | write the tasks that can run now as task descriptions `DIR/.json` | +| `--emit-tasks DIR` | write the tasks that can run now as task descriptions `DIR/.json`; retain their layout and stored input dependencies in the depot | | `--why TASK` | one task: its status, reasons, and its key parts beside those of the previous run | Reasons: `new-input`, `input-changed`, `stage-version`, `atom-changed`, `params-changed`, `context-changed`, @@ -99,6 +99,10 @@ Reasons: `new-input`, `input-changed`, `stage-version`, `atom-changed`, `params- another run); unknown tasks are `pending` on a task, or on `fetch:` for a bundle whose content is not known yet. See [contracts.md](contracts.md#plan-nnnotesplan1). +Census planning writes a run record for its completed tasks. Emitted descriptions have a saved task layout whose +input references keep stored inputs reachable during collection. Interrupted pipeline runs also retain completed +task results in their run records. + ### depot run Runs task descriptions (`nnnotes.task/1`, as `plan --emit-tasks` writes them) into a depot: each is checked (its key @@ -107,6 +111,9 @@ locators (the depot, its downloads, a file, or with `--fetch` the catalog), and `export assets` would store it. Several tasks run one after another in one process; stages that do not read bundles never load UnityPy. `--force` runs tasks whose result is stored again (the new result must be identical). +Each attempted task has a run record, selected by `depot.run` and its task id. The JSON report's `runs` array names +these records. Collection applies `--keep` separately to the recorded versions of each task. + ### catalog versions, import, fetch, diff The catalog versions of a depot: `import` a catalog file (with the APK's catalog from `[paths] apk` or diff --git a/docs/contracts.md b/docs/contracts.md index ed0bf6c..0e5f779 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -214,6 +214,10 @@ bytes nor keys; the file names of `original` and `cas` both follow from the reco `original` manifest alone. Documents derived for `original` from records and paths (the views with the files of their objects) are the exception: `original` has the derived documents, `cas` the artifacts they are made from. +Task-description layouts use the ordinary layout contract. Each task file entry may include `inputs`, an array of +content records (`sha256`, `size`) for the task's input dependencies. Collection follows those references as well as +the task-description file itself. + ## Product manifest: nnnotes.product/1 A product layout writes, for each product of `pipeline.assemble`, a manifest diff --git a/src/nnnotes/cli/assets.py b/src/nnnotes/cli/assets.py index 25f7c4d..fdb9915 100644 --- a/src/nnnotes/cli/assets.py +++ b/src/nnnotes/cli/assets.py @@ -1533,7 +1533,7 @@ def cmd_plan(args, cfg): if args.census: ws.run_index() facts = ws.facts(["catalog.index", "unity.census"], fetch=True) - Orchestrator( + run = Orchestrator( ws.depot, ws.scheduled, workers=ws.workers(), @@ -1543,10 +1543,16 @@ def cmd_plan(args, cfg): initializer=_worker_init, initargs=(cfg,), log=ws.log, - ).run(["catalog.index", "unity.census"], facts=facts) + ).run( + ["catalog.index", "unity.census"], + facts=facts, + context=ws.context(), + selection=["depot.plan.census", *ws.selection], + ) + run.write() p = ws.plan(pipeline, params, since=args.since, out=args.out, layouts=layouts) if args.emit_tasks: - files = p.emit(args.emit_tasks) + files = p.emit(args.emit_tasks, depot=ws.depot, context=ws.context(), selection=ws.selection) ws.log(f"{len(files)} task descriptions in {args.emit_tasks}") if args.why: common.report(args, why(ws, p, args.why, args.since), why_lines) @@ -1617,7 +1623,7 @@ def _task_files(items) -> list[Path]: def cmd_run_stage(args, cfg): from ..pipeline.depot import open_depot - from ..pipeline.orchestrate import run_one + from ..pipeline.orchestrate import Run, run_one problems, tasks = [], [] for f in _task_files(args.tasks): @@ -1644,7 +1650,7 @@ def cmd_run_stage(args, cfg): root = Session(cfg).depot_root() depot = open_depot(cfg, fetch=CatalogFetcher(root, cfg) if args.fetch else None) use(cfg) - out, bad = [], 0 + out, bad, runs = [], 0, [] for t in tasks: r = run_one(t, depot, inst.stages, args.force) if r["ok"]: @@ -1653,12 +1659,19 @@ def cmd_run_stage(args, cfg): else: out.append({"id": t.id, "key": t.key, "status": "failed", "code": r["code"], "message": r["message"]}) bad += 1 - if r["code"] == "config": - break + run = Run(depot, dict(t.context), ["depot.run", t.id], {t.stage: dict(t.params)}, {t.stage: t.version}) + run.tasks[t.id] = {k: out[-1][k] for k in ("id", "key", "status")} + runs.append(run.write()["id"]) + if not r["ok"] and r["code"] == "config": + break n = Counter(o["status"] for o in out) common.report( args, - {"tasks": out, "summary": {"tasks": len(tasks), "hit": n["hit"], "ran": n["ran"], "failed": n["failed"]}}, + { + "tasks": out, + "runs": runs, + "summary": {"tasks": len(tasks), "hit": n["hit"], "ran": n["ran"], "failed": n["failed"]}, + }, run_lines, ) if bad: diff --git a/src/nnnotes/pipeline/orchestrate.py b/src/nnnotes/pipeline/orchestrate.py index dd48d8c..eb973f4 100644 --- a/src/nnnotes/pipeline/orchestrate.py +++ b/src/nnnotes/pipeline/orchestrate.py @@ -590,8 +590,10 @@ def run( try: for name in pipeline: self._stage(self.stages[name], norm[name], env, force, run) - except ConfigError as e: - e.run = run # pyright: ignore[reportAttributeAccessIssue] (what ran before the stop: its results stay valid) + except BaseException as e: + if isinstance(e, ConfigError): + e.run = run # pyright: ignore[reportAttributeAccessIssue] + run.write() raise finally: self.close() @@ -813,6 +815,7 @@ def _record(self, run: Run, j: Job, sched: Scheduler, w) -> None: ) def _done(self, run: Run, t: Task, reply: dict, w) -> None: + run.tasks[t.id] = {"id": t.id, "key": t.key, "status": reply["status"] if reply["ok"] else "failed"} e = { "event": "end", "task": t.id, diff --git a/src/nnnotes/pipeline/plan.py b/src/nnnotes/pipeline/plan.py index e5184ea..7e8121e 100644 --- a/src/nnnotes/pipeline/plan.py +++ b/src/nnnotes/pipeline/plan.py @@ -28,7 +28,7 @@ from . import contract from .depot import write_file -from .orchestrate import WORKER_BYTES, jobs +from .orchestrate import WORKER_BYTES, Run, jobs from .stages import Env, Pending, Stage, describe @@ -76,15 +76,30 @@ def check(self) -> int: return 1 return int(any(v["write"] or v["remove"] for v in self.doc["summary"]["layouts"].values())) - def emit(self, directory) -> list[Path]: - """Write the runnable tasks as task descriptions /.json (one per distinct key).""" + def emit(self, directory, *, depot=None, context=None, selection=()) -> list[Path]: + """Write runnable tasks as /.json (one per key). With a depot, retain a task layout and + the stored inputs of these descriptions as a run root.""" d = Path(directory) d.mkdir(parents=True, exist_ok=True) - out = {} + out, entries = {}, [] for t in self.tasks: if t.key not in out: out[t.key] = d / f"{t.key}.json" - write_file(out[t.key], contract.encode(t.to_json())) + body = contract.encode(t.to_json()) + write_file(out[t.key], body) + if depot is not None: + entries.append( + { + "path": out[t.key].name, + "id": t.id, + "sha256": depot.put(body), + "size": len(body), + "inputs": [{"sha256": i.sha256, "size": i.size} for i in t.inputs], + } + ) + if depot is not None and entries: + layout = depot.put(contract.encode(contract.layout_doc("tasks", {}, entries))) + Run(depot, dict(context or {}), ["depot.plan.tasks", *selection], {}, {}).write({"tasks": layout}) return sorted(out.values()) def text(self, limit: int | None = 20) -> str: diff --git a/src/nnnotes/schemas/layout.schema.json b/src/nnnotes/schemas/layout.schema.json index 1d296f0..ec97a97 100644 --- a/src/nnnotes/schemas/layout.schema.json +++ b/src/nnnotes/schemas/layout.schema.json @@ -44,6 +44,18 @@ }, "id": { "type": "string" + }, + "inputs": { + "description": "Input content dependencies retained by an emitted task description.", + "type": "array", + "items": { + "type": "object", + "required": ["sha256", "size"], + "properties": { + "sha256": {"$ref": "urn:nnnotes:schema:common#/$defs/sha256"}, + "size": {"type": "integer", "minimum": 0} + } + } } } } diff --git a/tests/cli/test_assets.py b/tests/cli/test_assets.py index 1f6175c..ab15a5e 100644 --- a/tests/cli/test_assets.py +++ b/tests/cli/test_assets.py @@ -552,12 +552,16 @@ def test_plan_emit_tasks_and_run_stage_give_the_exported_depot(tmp_path, capsys, doc = json.loads(out) if not any(n["status"] == "run" for n in doc["nodes"]): break + Depot(tmp_path / "q").gc() code, out, err = nn(capsys, *flags(d), "depot", "run", tasks, "--depot", tmp_path / "q") assert code in (0, 1), err + Depot(tmp_path / "q").gc() rounds += 1 assert rounds == 3 # index + census; scripts + addresses; export skip = ["inputs", "runs", "costs", "catalogs", "downloads"] - assert task_tree(tmp_path / "orch", skip) == task_tree(tmp_path / "q", skip) + expected, actual = task_tree(tmp_path / "orch", skip), task_tree(tmp_path / "q", skip) + # The task runner also retains emitted descriptions and their input references. + assert {p: actual.get(p) for p in expected} == expected def test_plan_explains_and_waits_for_bundles_not_fetched(tmp_path, capsys, fake): @@ -1224,6 +1228,11 @@ def test_plan_census_takes_the_censuses_first(tmp_path, capsys, fake): census = [n for n in nodes if n["stage"] == "unity.census"] assert census and all(n["status"] == "hit" for n in census) assert {n["status"] for n in nodes if n["stage"] == "link.scripts"} == {"run"} + depot = Depot(tmp_path / "s") + run = depot.latest_run(["depot.plan.census", "all"]) + assert run and {t["id"] for t in run["tasks"]} >= {n["id"] for n in census} + depot.gc(keep=1) + assert all(depot.has_result(n["key"]) for n in census) class NamedByExport(Named): diff --git a/tests/pipeline/test_orchestrate.py b/tests/pipeline/test_orchestrate.py index 4b3533e..1b61955 100644 --- a/tests/pipeline/test_orchestrate.py +++ b/tests/pipeline/test_orchestrate.py @@ -304,6 +304,10 @@ def interrupt(*args): with pytest.raises(KeyboardInterrupt): orch.run(["toy.src"], facts={"files": toy_files(tmp_path / "in", {"a1": "x", "a2": "y"})}) assert seen == [] and orch._pool == [] + saved = depot.latest_run([]) + assert saved and any(t["status"] == "ran" for t in saved["tasks"]) + depot.gc(keep=1) + assert all(depot.has_result(t["key"]) for t in saved["tasks"] if t["key"]) def test_configuration_stop_keeps_the_incomplete_batch(tmp_path): @@ -323,6 +327,11 @@ def run(self, task, depot): orch.run(["toy.src"], facts={"files": toy_files(tmp_path / "in", {"a1": "x", "a2": "y", "a3": "z"})}) assert seen == [] + saved = orch.depot.latest_run([]) + assert saved and any(t["status"] == "ran" for t in saved["tasks"]) + orch.depot.gc(keep=1) + assert all(orch.depot.has_result(t["key"]) for t in saved["tasks"] if t["status"] == "ran") + def test_jobs_group_a_stages_batches(): def task(stage, name, cpu, peak): diff --git a/tests/pipeline/test_plan.py b/tests/pipeline/test_plan.py index 76cb105..f98693d 100644 --- a/tests/pipeline/test_plan.py +++ b/tests/pipeline/test_plan.py @@ -2,11 +2,14 @@ from pathlib import Path +import pytest + from nnnotes.pipeline import contract -from nnnotes.pipeline.contract import Task +from nnnotes.pipeline.contract import Input, Task from nnnotes.pipeline.depot import Depot from nnnotes.pipeline.orchestrate import Orchestrator -from nnnotes.pipeline.plan import plan, why +from nnnotes.pipeline.plan import Plan, plan, why +from nnnotes.pipeline.remote import open_remote from nnnotes.pipeline.stages import registry from tests.pipeline.test_stages import PIPELINE, Count, Src, Total, toy_files, toy_stages @@ -149,6 +152,31 @@ def test_emitted_tasks_are_runnable_descriptions(tmp_path): assert f.read_bytes() == contract.encode(t.to_json()) +@pytest.mark.parametrize("shared", [False, True]) +def test_emitted_tasks_and_stored_inputs_survive_collection(tmp_path, shared): + remote = open_remote((tmp_path / "remote").as_uri()) if shared else None + depot = Depot(tmp_path / "depot", remote=remote) + sha = depot.put(b"input") + unused = depot.put(b"unreferenced") + task = Task("toy.src", 1, "a", inputs=(Input("source", sha, 5, locators=({"kind": "depot"},)),)) + files = Plan({}, [task]).emit(tmp_path / "tasks", depot=depot, selection=["example"]) + run = depot.latest_run(["depot.plan.tasks", "example"]) + from tests.pipeline.test_contract import validators + + schemas = validators() + assert not list(schemas[contract.RUN].iter_errors(run)) + layout = depot.document(run["layouts"]["tasks"]) + assert not list(schemas[contract.LAYOUT].iter_errors(layout)) + depot.gc(keep=1, remote=shared, grace=0) + assert not depot.has(unused) + reader = Depot(tmp_path / "fresh", remote=remote) if shared else depot + saved = reader.document(reader.latest_run(["depot.plan.tasks", "example"])["layouts"]["tasks"]) + body = reader.read(saved["entries"][0]["sha256"]) + assert body == files[0].read_bytes() + restored = Task.from_json(contract.loads(body)) + assert reader.input_bytes(restored.inputs[0]) == b"input" + + def test_peak_estimate_oversize_and_layout_counts(tmp_path): depot, facts = setup(tmp_path, {"a": "x" * 100, "b": "y" * 10, "c": "z" * 5}) # peaks 1000, 100, 50 p = plan( From f4e3e8e1d70bbc3e2ba9551a99e9bcebf36db54e Mon Sep 17 00:00:00 2001 From: nichinichisou Date: Sun, 18 Oct 2026 20:27:58 +0800 Subject: [PATCH 2/2] fix(depot): publish stored inputs with emitted tasks --- src/nnnotes/pipeline/plan.py | 5 +++++ tests/pipeline/test_plan.py | 3 ++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/nnnotes/pipeline/plan.py b/src/nnnotes/pipeline/plan.py index 7e8121e..82869ca 100644 --- a/src/nnnotes/pipeline/plan.py +++ b/src/nnnotes/pipeline/plan.py @@ -88,6 +88,11 @@ def emit(self, directory, *, depot=None, context=None, selection=()) -> list[Pat body = contract.encode(t.to_json()) write_file(out[t.key], body) if depot is not None: + if depot.remote is not None: + for inp in t.inputs: + path = depot.path(inp.sha256) + if path.is_file() and depot.put_file(path) != (inp.sha256, inp.size): + raise ValueError(f"task {t.id}: stored input {inp.role} differs from its identity") entries.append( { "path": out[t.key].name, diff --git a/tests/pipeline/test_plan.py b/tests/pipeline/test_plan.py index f98693d..6ab42c8 100644 --- a/tests/pipeline/test_plan.py +++ b/tests/pipeline/test_plan.py @@ -155,9 +155,10 @@ def test_emitted_tasks_are_runnable_descriptions(tmp_path): @pytest.mark.parametrize("shared", [False, True]) def test_emitted_tasks_and_stored_inputs_survive_collection(tmp_path, shared): remote = open_remote((tmp_path / "remote").as_uri()) if shared else None - depot = Depot(tmp_path / "depot", remote=remote) + depot = Depot(tmp_path / "depot") sha = depot.put(b"input") unused = depot.put(b"unreferenced") + depot.remote = remote task = Task("toy.src", 1, "a", inputs=(Input("source", sha, 5, locators=({"kind": "depot"},)),)) files = Plan({}, [task]).emit(tmp_path / "tasks", depot=depot, selection=["example"]) run = depot.latest_run(["depot.plan.tasks", "example"])