fix(depot): retain planned and standalone task results

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.

docs/assets.md

@@ -91,14 +91,18 @@ 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/<key>.json` |
+| `--emit-tasks DIR` | write the tasks that can run now as task descriptions `DIR/<key>.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`,
 `previous-failed`, `result-missing`, `forced`; hits are `unchanged` or `cached` (the result is in the depot from
 another run); unknown tasks are `pending` on a task, or on `fetch:<file>` 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

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

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:

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,

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,35 @@ 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 <directory>/<key>.json (one per distinct key)."""
+    def emit(self, directory, *, depot=None, context=None, selection=()) -> list[Path]:
+        """Write runnable tasks as <directory>/<key>.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:
+                    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,
+                            "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:

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}
+              }
+            }
           }
         }
       }

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):

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):

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,32 @@ 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")
+    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"])
+    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(