feat(task_persist): close PARTIAL gaps — dichotomy wave

docs/dichotomy-rebase-honest-audit.md

@@ -2,14 +2,14 @@
 
 _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL only when 80%+ of its Python functions have name-preserving counterparts in the Go/Rust implementation — directories without the functions are facades and count as NOT TOUCHED._
 
-**FULL completion: 60%** (116/192 runtime modules at FULL)
+**FULL completion: 61%** (117/192 runtime modules at FULL)
 
-Function-level detail: 73% (1669/2288 runtime functions matched) — secondary; includes in-flight work
+Function-level detail: 73% (1674/2288 runtime functions matched) — secondary; includes in-flight work
 
 | Status | Modules | Lines | Meaning |
 |---|---|---|---|
-| FULL | 116 | 48317 | 80%+ of functions ported |
-| PARTIAL | 37 | 15677 | 20–80% — real logic exists, gaps named below |
+| FULL | 117 | 48586 | 80%+ of functions ported |
+| PARTIAL | 36 | 15408 | 20–80% — real logic exists, gaps named below |
 | NOT TOUCHED | 39 | 11471 | no name-preserving implementation |
 | Total runtime | 192 | 75465 | |
 
@@ -71,7 +71,6 @@ Function-level detail: 73% (1669/2288 runtime functions matched) — secondary;
 | core_server | 234 | 6/12 | make_server, log_message, _send_json, _send_ndjson, do_GET, do_POST |
 | voicebox | 164 | 5/11 | _run, start_recording, stop_recording, _transcribe_and_type, on_press, on_release |
 | persona_definition | 210 | 2/7 | _by_topic, compile_full, emit_constitution, emit_identity, emit_persona |
-| task_persist | 269 | 5/10 | _save, export, restore, main, handle_hook |
 | drift_observability | 399 | 5/9 | _do_core, main, _load_alerts, _save_alerts |
 | pins | 282 | 7/11 | _store_path, _detail_blob, unpin, list_pins |
 | prompt_sync | 487 | 11/15 | _now, _rust_available, get_store, main |
@@ -184,6 +183,7 @@ Function-level detail: 73% (1669/2288 runtime functions matched) — secondary;
 | shard_manager | 276 | 9/9 | RUST |
 | auto_context | 273 | 8/8 | GO |
 | spine_language | 269 | 5/5 | RUST |
+| task_persist | 269 | 10/10 | RUST |
 | construction_patterns_builtin | 266 | 1/1 | RUST |
 | iteration | 261 | 7/8 | RUST |
 | model_registry | 253 | 8/8 | GO |

scripts/check_task_persist_parity.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""Live parity: the task_persist gap-closing port vs the frozen truth
+(dichotomy wave #2479, worktree rune-wave-cw2).
+
+Composition law: the frozen truth was generated against the LIVE Python
+oracle and re-proved byte-identical by gen_task_persist_gap_truth.py
+--verify (live oracle == frozen); this script replays every truth row
+through the BUILT tp_parity binary — the same tpops engine the frozen
+battery replays in-process — and compares got vs expected exactly
+(port == frozen). Port == frozen == live oracle at 100%.
+
+Run:    python3 scripts/check_task_persist_parity.py
+Needs:  cargo build --release in src/rust/task_persist (tp_parity bin).
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+TRUTH = REPO / "src" / "rust" / "task_persist" / "src" / "frozen_tp_truth.json"
+BIN = (
+    REPO
+    / "src"
+    / "rust"
+    / "task_persist"
+    / "target"
+    / "release"
+    / "tp_parity"
+)
+
+
+def main() -> int:
+    truth = json.loads(TRUTH.read_text())
+    cases = truth["cases"]
+    proc = subprocess.run(
+        [str(BIN), str(TRUTH)],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    lines = [json.loads(ln) for ln in proc.stdout.splitlines() if ln.strip()]
+    if len(lines) != len(cases):
+        print(
+            f"FAIL: driver emitted {len(lines)} rows for {len(cases)} truth cases"
+        )
+        return 1
+
+    per_fn: dict[str, list[int]] = {}
+    fails: list[str] = []
+    for case, line in zip(cases, lines):
+        fn = case["fn"]
+        ok = line["got"] == case["expected"]
+        bucket = per_fn.setdefault(fn.rsplit(".", 1)[-1], [0, 0])
+        bucket[1] += 1
+        if ok:
+            bucket[0] += 1
+        else:
+            fails.append(fn)
+            print(f"DIVERGE [{fn}]")
+            print(f"  want: {json.dumps(case['expected'], sort_keys=True)[:400]}")
+            print(f"  got:  {json.dumps(line['got'], sort_keys=True)[:400]}")
+
+    total = len(cases)
+    passed = total - len(fails)
+    for fn in sorted(per_fn):
+        hit, n = per_fn[fn]
+        print(f"  {fn:<16} {hit}/{n}")
+    print(f"PARITY {passed}/{total} ({100.0 * passed / total:.1f}%)")
+    if fails:
+        print(f"FAILED FNS: {sorted(set(f.rsplit('.', 1)[-1] for f in fails))}")
+        return 1
+    print("OK: 100% parity with frozen truth (== live oracle)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/gen_task_persist_gap_truth.py

@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+"""gen_task_persist_gap_truth.py — task_persist gap-closing truth battery
+(dichotomy wave #2479, worktree rune-wave-cw2).
+
+Closes the PARTIAL audit gaps (_save, export, restore, main, handle_hook)
+and pins every branch of all 10 fns while at it.
+
+Uses the SHARED harness (scripts/freeze_truth.py) with the seams rig
+scripts/tp_freeze_seams/sitecustomize.py (auto-imported via PYTHONPATH by
+`freeze_truth.py --verify` too): the oracle's fns read module globals and
+print, so the rig wraps them in spec-driven sandbox replays — per-case
+fresh dirs, pre-state in the case args, aware frozen clock, COLUMNS=80,
+crash classes as {"raise": str(e)} with sandbox paths normalized to <sd>,
+and the written-file byte state ("raw" + "keys") riding every row that
+can touch the ledger. No wall-clock values, no absolute paths in truth.
+
+  python3 scripts/gen_task_persist_gap_truth.py            (freeze)
+  python3 scripts/gen_task_persist_gap_truth.py --verify   (re-freeze and
+      byte-compare against src/rust/task_persist/src/frozen_tp_truth.json)
+  PYTHONPATH=src:scripts/tp_freeze_seams python3 scripts/freeze_truth.py \
+      --verify src/rust/task_persist/src/frozen_tp_truth.json \
+      --module rune.task_persist                          (harness verify)
+"""
+
+import json
+import os
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+TARGET = REPO / "src" / "rust" / "task_persist" / "src" / "frozen_tp_truth.json"
+VERIFY = "--verify" in sys.argv
+
+sys.path.insert(0, str(REPO / "scripts"))
+sys.path.insert(0, str(REPO / "src"))
+
+# apply the seams rig explicitly (running this script directly does not
+# auto-import sitecustomize; PYTHONPATH runs do, and this import is cached)
+import importlib.util as _ilu  # noqa: E402
+
+_RIG = REPO / "scripts" / "tp_freeze_seams" / "sitecustomize.py"
+_spec = _ilu.spec_from_file_location("tp_freeze_seams_rig", str(_RIG))
+_rig = _ilu.module_from_spec(_spec)
+sys.modules["tp_freeze_seams_rig"] = _rig
+_spec.loader.exec_module(_rig)
+
+from freeze_truth import freeze, write_envelope  # noqa: E402
+
+FROZEN_NOW = _rig.FROZEN_NOW
+ORACLE = "rune.task_persist"
+
+
+# ── fixture task dicts (literal constants — no clocks, no paths) ─────────
+
+CREATED = "2026-10-01T00:00:00+00:00"
+
+
+def task(tid, summary, status="open", parent=None):
+    d = {"id": tid, "summary": summary, "status": status, "created": CREATED}
+    if parent is not None:
+        d["parent_id"] = parent
+    return d
+
+
+def plain(tid, summary, **extra):
+    d = {"id": tid, "summary": summary, "status": "open", "created": CREATED}
+    d.update(extra)
+    return d
+
+
+PRE_TWO_TASKS = json.dumps(
+    {"tasks": [task("T1", "ship the port", "in_progress"),
+               task("T2", "verify battery", parent="T1")],
+     "updated": "2026-10-02T09:30:00+00:00"},
+    indent=2) + "\n"
+
+PRE_WEIRD_IDS = json.dumps(
+    {"tasks": [
+        plain("T2", "two"), plain("T7", "seven"), plain("tt", "not T-prefixed"),
+        plain("Tx", "not an int"), plain("T-3", "negative"),
+        plain("T7x", "suffix junk"), plain("T3_2", "underscore junk"),
+    ], "updated": "2026-10-02T09:30:00+00:00"},
+    indent=2) + "\n"
+
+# int id isolated: the auto-id scan crashes on the FIRST truthy non-string
+# id — this row pins that crash class without masking the auto-id math
+PRE_INT_ID = json.dumps(
+    {"tasks": [plain(5, "int id")],
+     "updated": "2026-10-02T09:30:00+00:00"},
+    indent=2) + "\n"
+
+PRE_EVENTS = json.dumps(
+    {"tasks": [
+        {"id": "T1", "summary": "one", "status": "open", "created": CREATED,
+         "events": [{"time": "2026-10-01T01:00:00+00:00",
+                     "event": "created"}]},
+    ], "updated": "2026-10-01T02:00:00+00:00"},
+    indent=2) + "\n"
+
+PRE_LONG_SUMMARY = json.dumps(
+    {"tasks": [
+        plain("T1", "é" * 45 + "中" * 50 + "-tail-must-not-appear"),
+    ], "updated": "2026-10-01T02:00:00+00:00"},
+    indent=2) + "\n"
+
+PRE_EVENTS_NULL = json.dumps(
+    {"tasks": [
+        {"id": "T1", "summary": "one", "status": "open", "created": CREATED,
+         "events": None},
+    ], "updated": "2026-10-01T02:00:00+00:00"},
+    indent=2) + "\n"
+
+
+# ── case matrix ──────────────────────────────────────────────────────────
+
+def build_cases():
+    import copy
+    cases = []
+    add = lambda fn, spec: cases.append({"fn": fn, "args": [spec]})  # noqa: E731
+
+    # ── _load / load — default, corrupt(recovered), valid, odd shapes ────
+    add("_tp_load", {"pre": None})
+    add("_tp_load", {"pre": "{bad json"})
+    add("_tp_load", {"pre": PRE_TWO_TASKS})
+    add("_tp_load", {"pre": "{}"})
+    add("_tp_load", {"pre": "[1, 2]"})
+    add("_tp_load", {"pre": '{"tasks": [], "tasks": ["kept-later"]}'})
+    add("_tp_loadpub", {"pre": PRE_TWO_TASKS})
+
+    # ── _save — updated position pins + unicode bytes + crash class ──────
+    add("_tp_save", {"doc": {"tasks": [task("T1", "one")], "updated": None}})
+    add("_tp_save", {"doc": {"updated": "OLD-IN-PLACE",
+                             "tasks": [task("T1", "one")]}})
+    add("_tp_save", {"doc": {"tasks": [task("T2", "two")]}})
+    add("_tp_save", {"doc": {"tasks": [task("u1", "café 中 ✓ unicode")],
+                             "updated": None}})
+    add("_tp_save", {"doc": [1, 2]})
+
+    # ── export — replace/wipe/parse classes ──────────────────────────────
+    add("_tp_export", {"pre": None,
+                       "raw": [task("T1", "fresh one"), task("T2", "fresh two")]})
+    add("_tp_export", {"pre": PRE_TWO_TASKS, "raw": [task("T9", "replaced all")]})
+    add("_tp_export", {"pre": PRE_TWO_TASKS, "raw": []})
+    add("_tp_export", {"pre": None, "raw": []})
+    add("_tp_export", {"pre": PRE_TWO_TASKS,
+                       "raw": copy.deepcopy([task("T1", "x"), task("T2", "y")])})
+    add("_tp_export", {"pre": None, "raw": ""})
+    add("_tp_export", {"pre": None, "raw": "not-json"})
+    add("_tp_export", {"pre": None, "raw": "{bad"})
+    add("_tp_export", {"pre": None, "raw": '{"a": 1} x'})
+    add("_tp_export", {"pre": None, "raw": "[1 2]"})
+
+    # ── add — auto ids, explicit ids, truthy guards, crash classes ───────
+    add("_tp_add", {"pre": None, "summary": "ship the port"})
+    add("_tp_add", {"pre": PRE_WEIRD_IDS, "summary": "max plus one"})
+    add("_tp_add", {"pre": None, "summary": "explicit id", "id": "T42"})
+    add("_tp_add", {"pre": PRE_TWO_TASKS, "summary": "replaced row",
+                    "status": "in_progress", "id": "T1"})
+    add("_tp_add", {"pre": None, "summary": "child", "parent": "T0"})
+    add("_tp_add", {"pre": None, "summary": "blank parent", "parent": ""})
+    add("_tp_add", {"pre": None, "summary": "custom status",
+                    "status": "blocked"})
+    add("_tp_add", {"pre": None, "summary": "blank status", "status": ""})
+    add("_tp_add", {"pre": "{}", "summary": "keyerror doc"})
+    add("_tp_add", {"pre": PRE_INT_ID, "summary": "int id crash"})
+
+    # ── update — truthy guards, events setdefault, crash classes ─────────
+    add("_tp_update", {"pre": PRE_TWO_TASKS, "id": "T1", "status": "done",
+                       "summary": "shipped"})
+    add("_tp_update", {"pre": PRE_TWO_TASKS, "id": "T2", "event": "started"})
+    add("_tp_update", {"pre": PRE_EVENTS, "id": "T1", "event": "second event"})
+    add("_tp_update", {"pre": PRE_TWO_TASKS, "id": "T1", "status": "",
+                       "summary": "blank-status guard"})
+    add("_tp_update", {"pre": PRE_TWO_TASKS, "id": "T1"})
+    add("_tp_update", {"pre": PRE_TWO_TASKS, "id": "T99", "status": "done"})
+    add("_tp_update", {"pre": PRE_TWO_TASKS, "id": "T2", "event": ""})
+    add("_tp_update", {"pre": PRE_EVENTS_NULL, "id": "T1", "event": "boom"})
+    add("_tp_update", {"pre": "{}", "id": "T1", "status": "done"})
+    add("_tp_update", {"pre": PRE_WEIRD_IDS, "id": "T1"})
+
+    # ── restore — layout, defaults, update stamp, quote escape, crash ────
+    add("_tp_restore", {"pre": None})
+    add("_tp_restore", {"pre": PRE_TWO_TASKS})
+    add("_tp_restore", {"pre": '{"tasks": []}'})
+    add("_tp_restore", {"pre": '{"tasks": [{"id": "T1", "summary": '
+                              '"with \\"quotes\\"", "status": "in_progress"}]'
+                              '}'})
+    add("_tp_restore", {"pre": '{"tasks": [{"id": "T1"}]}'})
+    add("_tp_restore", {"pre": '{"updated": null, "tasks": [{"id": "T1"}]}'})
+    add("_tp_restore", {"pre": '{"tasks": [{"id": "T1", "summary": "p", '
+                              '"status": "open", "parent_id": ""}, '
+                              '{"id": "T2", "summary": "p", "status": '
+                              '"open", "parent_id": "T1"}]}'})
+    add("_tp_restore", {"pre": "[1, 2]"})
+
+    # ── status — header, rows, 80-CODEPOINT truncation, crash ────────────
+    add("_tp_status", {"pre": None})
+    add("_tp_status", {"pre": PRE_TWO_TASKS})
+    add("_tp_status", {"pre": PRE_LONG_SUMMARY})
+    add("_tp_status", {"pre": "[1, 2]"})
+
+    # ── main — argparse dispatch, both output modes, file state ──────────
+    add("_tp_main", {"pre": None, "argv": ["add", "--json", "ship it"]})
+    add("_tp_main", {"pre": None, "argv": ["add", "ship it"]})
+    add("_tp_main", {"pre": None, "argv": ["add"]})
+    add("_tp_main", {"pre": None, "argv": ["add", "blank status",
+                                           "--status", ""]})
+    add("_tp_main", {"pre": PRE_TWO_TASKS,
+                     "argv": ["update", "T1", "--status", "done"]})
+    add("_tp_main", {"pre": PRE_TWO_TASKS,
+                     "argv": ["--json", "update", "T99"]})
+    add("_tp_main", {"pre": None, "argv": ["update"]})
+    add("_tp_main", {"pre": None, "argv": ["export"],
+                     "stdin": json.dumps([task("T1", "from stdin"),
+                                          task("T2", "stdin two")])})
+    add("_tp_main", {"pre": None,
+                     "argv": ["export", "--json",
+                              json.dumps([task("T1", "arg JSON")])],
+                     "stdin": "ignored-arg-wins"})
+    add("_tp_main", {"pre": PRE_TWO_TASKS, "argv": ["export"], "stdin": "[]"})
+    add("_tp_main", {"pre": PRE_TWO_TASKS, "argv": ["status"]})
+    add("_tp_main", {"pre": PRE_TWO_TASKS, "argv": ["restore"]})
+    add("_tp_main", {"pre": None, "argv": ["bogus", "extra"]})
+    add("_tp_main", {"pre": None, "argv": []})
+    add("_tp_main", {"pre": None, "argv": ["--zz"]})
+    add("_tp_main", {"pre": PRE_WEIRD_IDS,
+                     "argv": ["--id", "T42", "add", "flagged id"]})
+
+    # ── handle_hook — session-end persist, echo, degrade classes ─────────
+    add("_tp_hook", {"pre": PRE_TWO_TASKS, "event": "on_session_end"})
+    add("_tp_hook", {"pre": None, "event": "on_session_end"})
+    add("_tp_hook", {"pre": PRE_TWO_TASKS, "event": "other_event"})
+    add("_tp_hook", {"pre": None, "event": "on_session_end",
+                     "blocked_dir": True})
+    add("_tp_hook", {"pre": "[1, 2]", "event": "on_session_end"})
+
+    return cases
+
+
+def refreeze():
+    return freeze(ORACLE, build_cases(), oracle_module=ORACLE)
+
+
+def main():
+    truth = refreeze()
+    write_envelope(FROZEN_NOW, truth, TARGET)
+    print(f"wrote {TARGET} ({len(truth)} cases)")
+
+
+def verify():
+    raw = json.loads(TARGET.read_text())
+    if raw.get("frozen_now"):
+        os.environ["RUNE_FROZEN_NOW"] = raw["frozen_now"]
+    frozen = raw["cases"]
+    fresh = refreeze()
+    if json.dumps(fresh, sort_keys=True) != json.dumps(frozen, sort_keys=True):
+        print(f"MISMATCH: regenerated truth differs from {TARGET}")
+        for f, g in zip(frozen, fresh):
+            if json.dumps(f, sort_keys=True) != json.dumps(g, sort_keys=True):
+                print(f"  first divergence at fn={f['fn']}")
+                print(f"  frozen:   {json.dumps(f['expected'])[:300]}")
+                print(f"  oracle:   {json.dumps(g['expected'])[:300]}")
+                break
+        return 1
+    print(f"OK: {len(frozen)} cases byte-identical against live oracle")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(verify() if VERIFY else main())

scripts/tp_freeze_seams/sitecustomize.py

@@ -0,0 +1,265 @@
+#!/usr/bin/env python3
+"""tp_freeze_seams — sitecustomize rig for freezing task_persist truth.
+
+Auto-imported (PYTHONPATH=scripts/tp_freeze_seams) by BOTH the generator
+(gen_task_persist_gap_truth.py) and `freeze_truth.py --verify --module
+rune.task_persist`, mirroring scripts/ia_freeze_seams. The oracle's fns
+read module globals (LEDGER_PATH/LOCK_PATH) and print — two places the
+shared harness cannot reach — so this rig replaces them with spec-driven
+wrappers that run the REAL oracle fn bodies:
+
+  * each `_tp_*` wrapper receives the case spec dict, seeds a FRESH
+    per-case sandbox from it (never the real ledger), repoints the
+    oracle's module globals into that sandbox, then calls the saved
+    oracle fn — the wrapper changes only what is RECORDED;
+  * print()-based fns (restore/status/main) run under redirect_stdout/
+    redirect_stderr and return the captured bytes as the frozen value;
+  * crash classes freeze as {"raise": str(e)} with sandbox paths
+    normalized to "<sd>" (no absolute paths in frozen truth);
+  * COLUMNS=80 pins argparse usage wrapping (terminal-width quirk);
+  * the module-global datetime is pinned to an AWARE frozen clock
+    (RUNE_FROZEN_NOW-compatible: freeze_truth's own naive-base pin
+    produces the same aware isoformat stamps and may replace it).
+
+The `_tp_*` names are frozen-truth machinery, NOT oracle fns — the
+dichotomy audit ignores underscore-prefixed non-oracle names on the
+Python side (it counts Python-defined fns in src/rune, which this rig
+never touches).
+"""
+import contextlib  # noqa: E402
+import copy  # noqa: E402
+import io  # noqa: E402
+import json  # noqa: E402
+import os  # noqa: E402
+import sys  # noqa: E402
+import tempfile  # noqa: E402
+from datetime import datetime as _dt, timezone as _tz  # noqa: E402
+from pathlib import Path  # noqa: E402
+
+os.environ["COLUMNS"] = "80"  # argparse usage wrap pin
+os.environ.setdefault(
+    "RUNE_MEMORY_DIR", tempfile.mkdtemp(prefix="tp_seams_import_")
+)  # import-time memory_env read lands in a throwaway dir
+
+_ROOT = Path(os.path.abspath(__file__)).parents[3]
+_SRC = str(_ROOT / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import rune.task_persist as _tp  # noqa: E402 — patched in place; freeze_truth's
+# import_module("rune.task_persist") resolves to THIS cached module object
+# (the oracle's own sys.path.insert(_SD) heals its `import memory_env`)
+
+# the REAL oracle fns, saved before any wrapping
+_REAL = {
+    "_load": _tp._load, "load": _tp.load, "_save": _tp._save,
+    "export": _tp.export, "add": _tp.add, "update": _tp.update,
+    "restore": _tp.restore, "status": _tp.status, "main": _tp.main,
+    "handle_hook": _tp.handle_hook,
+}
+
+FROZEN_NOW = "2026-10-15T08:00:00"
+_BASE = _dt.fromisoformat(FROZEN_NOW).replace(tzinfo=_tz.utc)
+
+
+class _Frozen(_dt):
+    @classmethod
+    def now(cls, tz=None):
+        if tz is not None:
+            return _BASE.replace(tzinfo=tz)
+        return _BASE
+
+
+_tp.datetime = _Frozen
+
+# ── sandbox plumbing (mirrors the generator's per-case dirs) ──────────────
+_SEQ = [0]
+_RUN_ROOT = Path(tempfile.mkdtemp(prefix="tp_seams_run_"))
+
+
+def _sandbox(blocked=False):
+    _SEQ[0] += 1
+    sd = _RUN_ROOT / f"case_{_SEQ[0]:04d}"
+    tasks_dir = sd / "tasks"
+    if blocked:
+        tasks_dir.parent.mkdir(parents=True, exist_ok=True)
+        tasks_dir.write_text("i am a file, not a directory\n")
+    else:
+        tasks_dir.mkdir(parents=True)
+    return sd, tasks_dir
+
+
+def _repoint(sd):
+    ledger = sd / "tasks" / "persisted.json"
+    _tp.LEDGER_DIR = str(sd / "tasks")
+    _tp.LEDGER_PATH = str(ledger)
+    _tp.LOCK_PATH = str(sd / "tasks" / ".lock")
+    return ledger
+
+
+def _seed(pre):
+    sd, _tasks_dir = _sandbox()
+    ledger = _repoint(sd)
+    if pre is not None:
+        ledger.write_text(pre)
+    return sd, ledger
+
+
+def _file_state(ledger):
+    if ledger is None or not ledger.exists():
+        return None, []
+    raw = ledger.read_text()
+    doc = json.loads(raw)
+    return raw, list(doc.keys()) if isinstance(doc, dict) else []
+
+
+def _norm(text, sd):
+    return text.replace(str(sd), "<sd>")
+
+
+def _norm_deep(value, sd):
+    if isinstance(value, dict):
+        return {k: _norm_deep(v, sd) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_norm_deep(v, sd) for v in value]
+    if isinstance(value, str):
+        return value.replace(str(sd), "<sd>")
+    return value
+
+
+def _run_err(fn):
+    """Call fn() ONCE; return (result, None) or (None, {"raise": str(e)})."""
+    box = {}
+    try:
+        def wrap():
+            box["res"] = fn()
+        wrap()
+        return box.get("res"), None
+    except BaseException as e:  # noqa: BLE001 — envelope is the contract
+        return None, {"raise": str(e)}
+
+
+# ── spec-driven wrappers (the frozen fn surface) ─────────────────────────
+
+def _tp_load(spec):
+    _seed(spec["pre"])
+    return {"result": copy.deepcopy(_REAL["_load"]())}
+
+
+def _tp_loadpub(spec):
+    _seed(spec["pre"])
+    return {"result": copy.deepcopy(_REAL["load"]())}
+
+
+def _tp_save(spec):
+    sd, _tasks_dir = _sandbox()
+    ledger = _repoint(sd)
+    res, err = _run_err(lambda: _REAL["_save"](copy.deepcopy(spec["doc"])))
+    if err:
+        return err
+    raw, keys = _file_state(ledger)
+    return {"result": copy.deepcopy(res), "raw": raw, "keys": keys}
+
+
+def _tp_export(spec):
+    sd, ledger = _seed(spec["pre"])
+    buf_err = io.StringIO()
+    with contextlib.redirect_stderr(buf_err):
+        res, err = _run_err(lambda: _REAL["export"](spec["raw"]))
+    if err:
+        err["stderr_norm"] = _norm(buf_err.getvalue(), sd)
+        return err
+    raw, keys = _file_state(ledger)
+    return {"result": copy.deepcopy(res), "raw": raw, "keys": keys,
+            "stderr_norm": _norm(buf_err.getvalue(), sd)}
+
+
+def _tp_add(spec):
+    sd, ledger = _seed(spec["pre"])
+    res, err = _run_err(lambda: _REAL["add"](
+        spec["summary"], spec.get("status", "open"),
+        spec.get("id"), spec.get("parent")))
+    if err:
+        return err
+    raw, keys = _file_state(ledger)
+    return {"result": copy.deepcopy(res), "raw": raw, "keys": keys,
+            "stderr_norm": ""}
+
+
+def _tp_update(spec):
+    sd, ledger = _seed(spec["pre"])
+    res, err = _run_err(lambda: _REAL["update"](
+        spec["id"], spec.get("status"), spec.get("summary"),
+        spec.get("event")))
+    if err:
+        return err
+    raw, keys = _file_state(ledger)
+    return {"result": copy.deepcopy(res), "raw": raw, "keys": keys,
+            "stderr_norm": ""}
+
+
+def _tp_restore(spec):
+    _seed(spec["pre"])
+    buf_out = io.StringIO()
+    with contextlib.redirect_stdout(buf_out):
+        _, err = _run_err(lambda: _REAL["restore"]())
+    text = buf_out.getvalue()
+    if err:
+        return {"stdout": text, "raise": err["raise"]}
+    return {"result": text}
+
+
+def _tp_status(spec):
+    _seed(spec["pre"])
+    buf_out = io.StringIO()
+    with contextlib.redirect_stdout(buf_out):
+        _, err = _run_err(lambda: _REAL["status"]())
+    text = buf_out.getvalue()
+    if err:
+        return {"stdout": text, "raise": err["raise"]}
+    return {"result": text}
+
+
+def _tp_main(spec):
+    sd, ledger = _seed(spec["pre"])
+    argv = ["task_persist.py"] + list(spec["argv"])
+    old_argv, old_stdin = sys.argv, sys.stdin
+    sys.argv = argv
+    sys.stdin = io.StringIO(spec.get("stdin") or "")
+    buf_out, buf_err = io.StringIO(), io.StringIO()
+    code = 0
+    try:
+        with contextlib.redirect_stdout(buf_out), contextlib.redirect_stderr(buf_err):
+            try:
+                _REAL["main"]()
+            except SystemExit as e:
+                code = e.code if isinstance(e.code, int) else 0
+    finally:
+        sys.argv, sys.stdin = old_argv, old_stdin
+    raw, keys = _file_state(ledger)
+    return {"exit": code, "stdout": buf_out.getvalue(),
+            "stderr": buf_err.getvalue(), "raw": raw, "keys": keys}
+
+
+def _tp_hook(spec):
+    blocked = spec.get("blocked_dir", False)
+    if blocked:
+        sd, _tasks_dir = _sandbox(blocked=True)
+        ledger = _repoint(sd)
+        ledger = None  # no file state available on the makedirs crash
+    else:
+        sd, _tasks_dir = _sandbox()
+        ledger = _repoint(sd)
+        if spec["pre"] is not None:
+            ledger.write_text(spec["pre"])
+    res = copy.deepcopy(_REAL["handle_hook"](spec["event"]))
+    raw, keys = _file_state(ledger)
+    return {"result": _norm_deep(res, sd), "raw": raw, "keys": keys,
+            "stderr_norm": ""}
+
+
+for _name in (
+    "_tp_load", "_tp_loadpub", "_tp_save", "_tp_export", "_tp_add",
+    "_tp_update", "_tp_restore", "_tp_status", "_tp_main", "_tp_hook",
+):
+    setattr(_tp, _name, globals()[_name])

src/rust/task_persist/Cargo.lock

@@ -2,12 +2,40 @@
 # It is not intended for manual editing.
 version = 4
 
+[[package]]
+name = "equivalent"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826"
+
+[[package]]
+name = "hashbrown"
+version = "0.17.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+
+[[package]]
+name = "indexmap"
+version = "2.14.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
+dependencies = [
+ "equivalent",
+ "hashbrown",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
 
+[[package]]
+name = "libc"
+version = "0.2.190"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78"
+
 [[package]]
 name = "memchr"
 version = "2.8.3"
@@ -68,6 +96,7 @@ version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
+ "indexmap",
  "itoa",
  "memchr",
  "serde",
@@ -90,6 +119,7 @@ dependencies = [
 name = "task_persist"
 version = "0.1.0"
 dependencies = [
+ "libc",
  "serde",
  "serde_json",
 ]

src/rust/task_persist/Cargo.toml

@@ -8,4 +8,10 @@ crate-type = ["rlib", "cdylib", "staticlib"]
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+# preserve_order: Python-dict insertion-order semantics — the oracle's
+# ledger is json.load-ed raw and re-dumped (json.dump(data, indent=2));
+# task dicts are arbitrary JSON and key order must survive the round trip.
+serde_json = { version = "1", features = ["preserve_order"] }
+# flock: oracle add/update hold an fcntl.flock(LOCK_EX) on "<dir>/.lock"
+# for the whole read-modify-write; libc::flock is the 1:1 equivalent.
+libc = "0.2"

src/rust/task_persist/src/bin/tp_parity.rs

@@ -0,0 +1,59 @@
+//! tp_parity — live-parity driver: replays the frozen truth rows through
+//! the BUILT binary and prints one JSON line per row:
+//!   {"fn": <dotted>, "got": <envelope>}
+//! scripts/check_task_persist_parity.py compares got against the frozen
+//! expected (which --verify proved equal to the live oracle) — the
+//! composition pins port == live oracle.
+
+use serde_json::{Map, Value};
+
+use task_persist::tpops;
+
+static RUN: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    let truth_path = args.get(1).cloned().expect("usage: tp_parity <truth.json>");
+    let truth = std::fs::read_to_string(&truth_path).expect("truth reads");
+    let doc: Value = serde_json::from_str(&truth).expect("truth parses");
+    let run = RUN.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+    let work = std::env::temp_dir().join(format!("tp_parity_{}_{run}", std::process::id()));
+    std::fs::create_dir_all(&work).expect("work dir creates");
+
+    let now = format!(
+        "{}+00:00",
+        doc.get("frozen_now")
+            .and_then(Value::as_str)
+            .expect("frozen_now")
+    );
+    let out: std::io::Stdout = std::io::stdout();
+    let mut w = std::io::BufWriter::new(out.lock());
+    use std::io::Write;
+    for case in doc.get("cases").and_then(Value::as_array).expect("cases") {
+        let op = case
+            .get("fn")
+            .and_then(Value::as_str)
+            .expect("fn")
+            .rsplit('.')
+            .next()
+            .expect("op");
+        let spec = case
+            .get("args")
+            .and_then(Value::as_array)
+            .and_then(|a| a.first())
+            .cloned()
+            .expect("spec");
+        let got = tpops::replay_case(op, &spec, &work, &now);
+        let mut line = Map::new();
+        line.insert("fn".into(), case.get("fn").cloned().unwrap_or(Value::Null));
+        line.insert("got".into(), got);
+        line.insert(
+            "expected".into(),
+            case.get("expected").cloned().unwrap_or(Value::Null),
+        );
+        let _ = serde_json::to_writer(&mut w, &Value::Object(line));
+        let _ = w.write_all(b"\n");
+        let _ = w.flush();
+    }
+    let _ = std::fs::remove_dir_all(&work);
+}

src/rust/task_persist/src/frozen_tp_battery.rs

@@ -0,0 +1,80 @@
+//! frozen_tp_battery — the frozen-truth battery for the task_persist
+//! gap-close (dichotomy wave #2479). Replays scripts/
+//! gen_task_persist_gap_truth.py output VERBATIM (include_str!) — the
+//! truth is the LIVE oracle's frozen behavior (gen×2 byte-identical,
+//! harness --verify OK ×2 against the live oracle); the battery only
+//! interprets the row spec. Sandbox paths normalize to "<sd>" before the
+//! compare (the oracle truth carries them that way — no abs-path law).
+
+use serde_json::Value;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use super::tpops;
+
+const TRUTH: &str = include_str!("frozen_tp_truth.json");
+
+static RUN: AtomicUsize = AtomicUsize::new(0);
+
+#[test]
+fn frozen_truth_replays() {
+    let doc: Value = serde_json::from_str(TRUTH).expect("frozen_tp_truth.json parses");
+    let cases = doc
+        .get("cases")
+        .and_then(Value::as_array)
+        .expect("cases array")
+        .clone();
+    assert!(!cases.is_empty(), "truth must not be empty");
+
+    let run = RUN.fetch_add(1, Ordering::SeqCst);
+    let work = if run > 0 {
+        std::env::temp_dir().join(format!("tp_batt_{}_{run}", std::process::id()))
+    } else {
+        std::env::temp_dir().join(format!("tp_batt_{}", std::process::id()))
+    };
+    std::fs::create_dir_all(&work).expect("work dir creates");
+
+    let now = format!(
+        "{}+00:00",
+        doc.get("frozen_now")
+            .and_then(Value::as_str)
+            .expect("frozen_now")
+    );
+
+    let mut failures: Vec<(String, Value, Value)> = Vec::new();
+    for case in &cases {
+        let op = case
+            .get("fn")
+            .and_then(Value::as_str)
+            .expect("fn name")
+            .rsplit('.')
+            .next()
+            .expect("op suffix")
+            .to_string();
+        let spec = case
+            .get("args")
+            .and_then(Value::as_array)
+            .and_then(|a| a.first())
+            .cloned()
+            .expect("spec dict");
+        let expected = case.get("expected").cloned().expect("expected envelope");
+        let got = tpops::replay_case(&op, &spec, &work, &now);
+        if got != expected {
+            failures.push((op, expected, got));
+        }
+    }
+    let _ = std::fs::remove_dir_all(&work);
+    for (op, want, got) in &failures {
+        eprintln!("BATTERY DIVERGE [{op}]");
+        eprintln!(
+            "  want: {}",
+            serde_json::to_string(want).unwrap_or_default()
+        );
+        eprintln!("  got:  {}", serde_json::to_string(got).unwrap_or_default());
+    }
+    assert!(
+        failures.is_empty(),
+        "frozen truth battery: {}/{} rows diverged",
+        failures.len(),
+        cases.len()
+    );
+}

src/rust/task_persist/src/tpops.rs

@@ -0,0 +1,255 @@
+//! tpops — the truth-op engine shared by the frozen battery and the
+//! tp_parity driver: seeds the per-case sandbox from the row's spec,
+//! calls the ported fn exactly as the oracle shim did, and returns the
+//! observed envelope (normalized: sandbox paths -> "<sd>").
+
+use crate::{_load, _save, add, export, handle_hook, restore, status, update, MainOutcome};
+use serde_json::{Map, Number, Value};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+static SEQ: AtomicUsize = AtomicUsize::new(0);
+
+/// One truth row replay. `work` is the run root; the case sandbox is
+/// `<work>/case_NNNN` (fresh dir per row, cleaned by the caller).
+pub fn replay_case(op: &str, spec: &Value, work: &Path, now: &str) -> Value {
+    let n = SEQ.fetch_add(1, Ordering::SeqCst);
+    let case_dir: PathBuf = work.join(format!("case_{n:04}"));
+    let tasks_dir = case_dir.join("tasks");
+
+    let blocked = spec
+        .get("blocked_dir")
+        .and_then(Value::as_bool)
+        .unwrap_or(false);
+
+    let ledger = if blocked {
+        // the shim blocks the TASKS DIR NAME (the makedirs target is a
+        // file — the FileExistsError class pin); ledger paths mirror the
+        // oracle shim's layout
+        std::fs::create_dir_all(&case_dir).expect("case dir creates");
+        std::fs::write(&tasks_dir, "file, not dir\n").expect("blocked file");
+        case_dir.join("tasks/persisted.json")
+    } else {
+        std::fs::create_dir_all(&tasks_dir).expect("case dir creates");
+        case_dir.join("tasks/persisted.json")
+    };
+    if let Some(pre) = spec.get("pre").and_then(Value::as_str) {
+        std::fs::write(&ledger, pre).expect("pre seeds");
+    }
+
+    let got = run_op(op, spec, &ledger, now);
+    // sandbox cleanup then normalize (crash messages embed the case dir)
+    let _ = std::fs::remove_dir_all(&case_dir);
+    let needle = case_dir.to_string_lossy().into_owned();
+    norm_deep(got, &needle)
+}
+
+fn norm_deep(v: Value, needle: &str) -> Value {
+    match v {
+        Value::String(s) => Value::String(s.replace(needle, "<sd>")),
+        Value::Array(a) => Value::Array(a.into_iter().map(|x| norm_deep(x, needle)).collect()),
+        Value::Object(m) => Value::Object(
+            m.into_iter()
+                .map(|(k, x)| (k, norm_deep(x, needle)))
+                .collect(),
+        ),
+        other => other,
+    }
+}
+
+fn run_op(op: &str, spec: &Value, ledger: &Path, now: &str) -> Value {
+    match op {
+        "_tp_load" | "_tp_loadpub" => match _load(ledger) {
+            Ok(doc) => env4("result", doc),
+            Err(msg) => raise(&msg),
+        },
+        "_tp_save" => {
+            let mut doc = spec.get("doc").cloned().unwrap_or(Value::Null);
+            match _save(&mut doc, ledger, now) {
+                Ok(()) => {
+                    let (raw, keys) = file_state(ledger);
+                    let mut m = Map::new();
+                    m.insert("result".into(), Value::Null);
+                    m.insert("raw".into(), raw);
+                    m.insert("keys".into(), Value::Array(keys));
+                    Value::Object(m)
+                }
+                Err(msg) => raise(&msg),
+            }
+        }
+        "_tp_export" => {
+            let sink = WarnSink::default();
+            let raw_arg = spec.get("raw").cloned().unwrap_or(Value::Null);
+            let mut emit = |msg: String| sink.push(&msg);
+            match export(ledger, &raw_arg, now, &mut emit) {
+                Ok(res) => {
+                    let (raw, keys) = file_state(ledger);
+                    let mut m = Map::new();
+                    m.insert("result".into(), res);
+                    m.insert("raw".into(), raw);
+                    m.insert("keys".into(), Value::Array(keys));
+                    m.insert("stderr_norm".into(), Value::String(sink.text()));
+                    Value::Object(m)
+                }
+                Err(msg) => {
+                    let mut m = Map::new();
+                    m.insert("raise".into(), Value::String(msg));
+                    m.insert("stderr_norm".into(), Value::String(sink.text()));
+                    Value::Object(m)
+                }
+            }
+        }
+        "_tp_add" => io_op_result(
+            add(
+                ledger,
+                spec.get("summary").and_then(Value::as_str).unwrap_or(""),
+                spec.get("status").and_then(Value::as_str).unwrap_or("open"),
+                spec.get("id").and_then(Value::as_str),
+                spec.get("parent").and_then(Value::as_str),
+                now,
+            ),
+            ledger,
+        ),
+        "_tp_update" => io_op_result(
+            update(
+                ledger,
+                spec.get("id").and_then(Value::as_str).unwrap_or(""),
+                spec.get("status").and_then(Value::as_str),
+                spec.get("summary").and_then(Value::as_str),
+                spec.get("event").and_then(Value::as_str),
+                now,
+            ),
+            ledger,
+        ),
+        "_tp_restore" => text_op(ledger, restore),
+        "_tp_status" => text_op(ledger, status),
+        "_tp_main" => {
+            let argv: Vec<String> = std::iter::once("task_persist.py".to_string())
+                .chain(
+                    spec.get("argv")
+                        .and_then(Value::as_array)
+                        .into_iter()
+                        .flatten()
+                        .filter_map(|v| v.as_str().map(str::to_string)),
+                )
+                .collect();
+            let stdin = spec
+                .get("stdin")
+                .and_then(Value::as_str)
+                .map(str::to_string);
+            let sink = WarnSink::default();
+            let mut emit = |msg: String| sink.push(&msg);
+            let MainOutcome {
+                exit,
+                stdout,
+                stderr,
+            } = crate::main_with_sink(&argv, stdin.as_deref(), ledger, now, &mut emit);
+            let (raw, keys) = file_state(ledger);
+            let combined_stderr = if stderr.is_empty() {
+                sink.text()
+            } else {
+                format!("{stderr}{}", sink.text())
+            };
+            let mut m = Map::new();
+            m.insert("exit".into(), Value::Number(Number::from(exit as i64)));
+            m.insert("stdout".into(), Value::String(stdout));
+            m.insert("stderr".into(), Value::String(combined_stderr));
+            m.insert("raw".into(), raw);
+            m.insert("keys".into(), Value::Array(keys));
+            Value::Object(m)
+        }
+        "_tp_hook" => {
+            let event = spec.get("event").and_then(Value::as_str).unwrap_or("");
+            let res = handle_hook(ledger, event, now);
+            let (raw, keys) = file_state(ledger);
+            let mut m = Map::new();
+            m.insert("result".into(), res);
+            m.insert("raw".into(), raw);
+            m.insert("keys".into(), Value::Array(keys));
+            m.insert("stderr_norm".into(), Value::String(String::new()));
+            Value::Object(m)
+        }
+        _ => {
+            let mut m = Map::new();
+            m.insert("raise".into(), Value::String(format!("unknown op {op}")));
+            Value::Object(m)
+        }
+    }
+}
+
+fn raise(msg: &str) -> Value {
+    let mut m = Map::new();
+    m.insert("raise".into(), Value::String(msg.to_string()));
+    Value::Object(m)
+}
+
+fn env4(key: &str, v: Value) -> Value {
+    let mut m = Map::new();
+    m.insert(key.into(), v);
+    Value::Object(m)
+}
+
+fn io_op_result(res: Result<Value, String>, ledger: &Path) -> Value {
+    match res {
+        Ok(v) => {
+            let (raw, keys) = file_state(ledger);
+            let mut m = Map::new();
+            m.insert("result".into(), v);
+            m.insert("raw".into(), raw);
+            m.insert("keys".into(), Value::Array(keys));
+            m.insert("stderr_norm".into(), Value::String(String::new()));
+            Value::Object(m)
+        }
+        Err(msg) => raise(&msg),
+    }
+}
+
+fn text_op(ledger: &Path, f: fn(&Value, &mut String) -> Result<(), String>) -> Value {
+    let doc = match _load(ledger) {
+        Ok(d) => d,
+        Err(msg) => return raise(&msg),
+    };
+    let mut out = String::new();
+    match f(&doc, &mut out) {
+        Ok(()) => env4("result", Value::String(out)),
+        Err(msg) => {
+            let mut m = Map::new();
+            m.insert("stdout".into(), Value::String(out));
+            m.insert("raise".into(), Value::String(msg));
+            Value::Object(m)
+        }
+    }
+}
+
+fn file_state(ledger: &Path) -> (Value, Vec<Value>) {
+    let text = match std::fs::read_to_string(ledger) {
+        Ok(t) => t,
+        Err(_) => return (Value::Null, vec![]),
+    };
+    match serde_json::from_str::<Value>(&text) {
+        Ok(v @ Value::Object(_)) => (
+            Value::String(text),
+            v.as_object()
+                .map(|m| m.keys().map(|k| k.clone().into()).collect())
+                .unwrap_or_default(),
+        ),
+        _ => (Value::String(text), vec![]),
+    }
+}
+
+#[derive(Default)]
+struct WarnSink {
+    parts: std::sync::Mutex<Vec<String>>,
+}
+
+impl WarnSink {
+    fn push(&self, msg: &str) {
+        self.parts
+            .lock()
+            .expect("sink lock")
+            .push(format!("{msg}\n"));
+    }
+    fn text(&self) -> String {
+        self.parts.lock().expect("sink lock").join("")
+    }
+}

tests/test_task_persist.py

@@ -0,0 +1,154 @@
+"""Tests: task_persist.py surface (dichotomy wave #2479, gap-close).
+
+The module has NO delegation seam (pure Python CLI/plugin module — the
+Rust port is a standalone crate driven by tp_parity), so the both-pins
+gate is: the module's behavior must be IDENTICAL with RUNE_TASK_PERSIST_CORE
+unset and set to "python" (the env the loader family would consult —
+task_persist ignores it by design; this pins that the env stays inert).
+
+Every test runs against a sandboxed RUNE_MEMORY_DIR — the real ledger
+under /mnt/workdrive/rune/memory is never touched.
+"""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "src" / "rune" / "task_persist.py"
+
+
+@pytest.fixture()
+def sandbox(monkeypatch):
+    sd = Path(os.environ.get("TMPDIR", "/tmp")) / "tp_test_sandbox"
+    tasks = sd / "tasks"
+    tasks.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setenv("RUNE_MEMORY_DIR", str(sd))
+    ledger = tasks / "persisted.json"
+    if ledger.exists():
+        ledger.unlink()
+    return {"sd": sd, "ledger": ledger}
+
+
+def cli(*argv, env_extra=None, stdin=None):
+    env = dict(os.environ)
+    if env_extra:
+        env.update(env_extra)
+    return subprocess.run(
+        [sys.executable, str(SCRIPT), *argv],
+        capture_output=True, text=True, env=env, input=stdin, timeout=60,
+    )
+
+
+def test_import_is_pure_python_surface(sandbox):
+    sys.path.insert(0, str(ROOT / "src"))
+    sys.path.insert(0, str(ROOT / "src" / "rune"))
+    for name in list(sys.modules):
+        if name == "task_persist":
+            del sys.modules[name]
+    env = os.environ.copy()
+    sys.path.insert(0, str(ROOT / "src/rune"))
+    # fresh import under the sandbox; the module reads env at import time
+    import importlib
+
+    task_persist = importlib.import_module("task_persist")
+    assert task_persist.LEDGER_DIR.endswith(os.path.join("tasks"))
+    assert task_persist.LEDGER_PATH.endswith("persisted.json")
+
+
+def test_load_default_shape(sandbox):
+    ledger = sandbox["ledger"]
+    data = json.loads(subprocess.run(
+        [sys.executable, "-c",
+         "import sys, json; sys.path.insert(0, %r); sys.path.insert(0, %r); "
+         "import task_persist; print(json.dumps(task_persist.load()))"
+         % (str(ROOT / "src"), str(SCRIPT.parent))],
+        capture_output=True, text=True, env=os.environ,
+    ).stdout)
+    assert data == {"tasks": [], "updated": None}
+
+
+def test_corrupted_file_recovers(sandbox):
+    sandbox["ledger"].write_text("{bad json")
+    data = json.loads(subprocess.run(
+        [sys.executable, "-c",
+         "import sys, json; sys.path.insert(0, %r); sys.path.insert(0, %r); "
+         "import task_persist; print(json.dumps(task_persist.load()))"
+         % (str(ROOT / "src"), str(SCRIPT.parent))],
+        capture_output=True, text=True, env=os.environ,
+    ).stdout)
+    assert data == {"tasks": [], "updated": None, "recovered": True}
+
+
+def test_add_update_restore_cli_roundtrip(sandbox):
+    r = cli("add", "ship the port")
+    assert r.returncode == 0 and "Added T1: ship the port" in r.stdout
+    r = cli("update", "T1", "--status", "done")
+    assert "Updated T1" in r.stdout
+    r = cli("status")
+    assert "[done        ] T1: ship the port" in r.stdout
+    r = cli("restore")
+    assert "# 1 persisted tasks (last updated: " in r.stdout
+    assert '[done        ] T1: ship the port' in r.stdout
+
+
+def test_export_replaces_and_warns_on_wipe(sandbox):
+    cli("add", "one")
+    cli("add", "two")
+    r = cli("export", json.dumps([{"id": "T9", "summary": "fresh"}]))
+    assert "Exported 1 tasks" in r.stdout
+    r = cli("export", "[]")
+    assert "Exported 0 tasks" in r.stdout
+    assert r.stderr.rstrip() == ("[task_persist] WARNING: export with empty "
+                                 "list will wipe 1 existing tasks")
+
+
+def test_add_usage_and_invalid_choice(sandbox):
+    r = cli("add")
+    assert r.returncode == 0
+    assert "usage: task_persist.py add '<summary>'" in r.stdout
+    r = cli("bogus")
+    assert r.returncode == 2
+    assert "error: argument cmd: invalid choice: 'bogus'" in r.stderr
+
+
+def test_parent_id_and_auto_id(sandbox):
+    cli("add", "parent task")
+    cli("add", "child", "--parent", "T1")
+    data = json.loads(sandbox["ledger"].read_text())
+    assert [t["id"] for t in data["tasks"]] == ["T1", "T2"]
+    assert data["tasks"][1]["parent_id"] == "T1"
+
+
+def test_env_pin_inert_both_ways(sandbox):
+    """Both pins: the CLI surface is IDENTICAL with the (hypothetical)
+    RUNE_TASK_PERSIST_CORE env unset vs pinned to python — task_persist
+    has no delegation seam; this proves the env stays inert."""
+    seeded = json.dumps({"tasks": [{"id": "T1", "summary": "pinned",
+                                    "status": "open"}]}, indent=2)
+    sandbox["ledger"].write_text(seeded)
+    out = []
+    for extra in ({}, {"RUNE_TASK_PERSIST_CORE": "python"}):
+        r1 = cli("status", env_extra=extra)
+        r2 = cli("restore", env_extra=extra)
+        r3 = cli("add", "two", env_extra=extra)
+        r4 = cli("add", "again", env_extra=extra)
+        ledger_after = sandbox["ledger"].read_text()
+        # wall-clock keys are NOT part of the both-pins contract (the frozen
+        # battery pins timestamps against the oracle clock); strip them
+        doc = json.loads(ledger_after)
+        for t in doc.get("tasks", []):
+            t.pop("created", None)
+            t.pop("updated", None)
+        doc.pop("updated", None)
+        out.append((r1.stdout, r2.stdout, r3.stdout, r4.stdout, doc))
+        # re-seed from the ORIGINAL constant (dropping whatever pin-1's
+        # writes stamped onto the doc)
+        sandbox["ledger"].write_text(seeded)
+    assert out[0] == out[1]
+    assert "[open        ] T1: pinned" in out[0][0]

verification/tp-cargo-build.exit

@@ -0,0 +1 @@
+exit=0

verification/tp-cargo-test.exit

@@ -0,0 +1 @@
+exit=0

verification/tp-clippy.exit

@@ -0,0 +1 @@
+exit=0

verification/tp-rustfmt.exit

@@ -0,0 +1 @@
+exit=0

verification/tp-summary.md

@@ -0,0 +1,50 @@
+# task_persist gap-close verification summary
+
+Module: task_persist (dichotomy wave #2479, worktree rune-wave-cw2, resume of cancelled run)
+Oracle: src/rune/task_persist.py (269L, 10 fns)
+Before: 5/10 PARTIAL (missing: _save, export, restore, main, handle_hook) → After: **10/10 FULL (RUST)**
+
+## What this run added over the cancelled mid-flight state
+- The cancelled run's truth was gen-script-only verifiable (shim module names the
+  shared harness cannot resolve in a fresh process). Rewritten onto the seams-rig
+  pattern (scripts/tp_freeze_seams/sitecustomize.py, ia_freeze_seams precedent):
+  truth rows now carry bare `_tp_*` fn names on `rune.task_persist`, so the
+  LITERAL standing-law command passes:
+  `PYTHONPATH=src:scripts/tp_freeze_seams python3 scripts/freeze_truth.py --verify src/rust/task_persist/src/frozen_tp_truth.json --module rune.task_persist`
+- crate-type restored to the fleet triple ["rlib","cdylib","staticlib"] (spine
+  convention; the mid-flight edit had rlib-only).
+- scripts/check_task_persist_parity.py created (referenced by the mid-flight
+  tp_parity driver but missing).
+- tests/test_task_persist.py (mid-flight, untracked) verified and committed.
+
+## Verification (all logs in this directory, tp- prefix)
+| gate | command | result | evidence |
+|---|---|---|---|
+| truth regen ×2 | `python3 scripts/gen_task_persist_gap_truth.py` ×2 | 75 cases, byte-identical | tp-truth-regen1.log, tp-truth-regen2.log |
+| gen verify | `python3 scripts/gen_task_persist_gap_truth.py --verify` | OK: 75 cases byte-identical | tp-freeze-verify.log |
+| harness verify (the law) | `PYTHONPATH=src:scripts/tp_freeze_seams python3 scripts/freeze_truth.py --verify src/rust/task_persist/src/frozen_tp_truth.json --module rune.task_persist` | OK: 75 cases byte-identical | tp-harness-verify.log |
+| cargo build | `cargo build --release --all-targets` (src/rust/task_persist) | exit 0 | tp-cargo-build.log |
+| frozen battery | `cargo test --release` | frozen_truth_replays ok (75 rows verbatim) | tp-cargo-test.log |
+| clippy | `cargo clippy --release --all-targets -- -D warnings` | exit 0, no warnings | tp-clippy.log |
+| rustfmt | `cargo fmt --check` | exit 0 | tp-rustfmt.log |
+| live parity | `python3 scripts/check_task_persist_parity.py` (tp_parity binary vs frozen truth) | PARITY 75/75 (100%) — all 10 fns | tp-live-parity.log |
+| pytest live pin | `python3 -m pytest tests/test_task_persist.py -q` | 8 passed | tp-pytest-live.log |
+| pytest python pin | `RUNE_TASK_PERSIST_CORE=python python3 -m pytest tests/test_task_persist.py -q` | 8 passed | tp-pytest-python.log |
+| audit | `PYTHONPATH=src python3 scripts/dichotomy_audit.py --write` | task_persist 5/10 → 10/10 RUST | tp-audit.log, docs/dichotomy-rebase-honest-audit.md |
+
+## Per-fn parity (frozen corpus)
+_tp_load 6/6 · _tp_loadpub 1/1 · _tp_save 5/5 · _tp_export 10/10 · _tp_add 10/10 ·
+_tp_update 10/10 · _tp_restore 8/8 · _tp_status 4/4 · _tp_main 16/16 · _tp_hook 5/5
+
+## Documented divergences (not bugs; outside the frozen corpus)
+- NaN/Infinity literals in ledger files: Python json accepts them; the port's
+  serde_json parse treats them as corrupt → recovered doc (lib.rs header).
+- argparse `-h/--help` render: not ported (usage/epilog help text is
+  machine-rendered argparse machinery).
+- json.JSONDecodeError message parity covers the frozen corpus classes
+  (Expecting value / property name / ',' delimiter / Extra data / control
+  character / escape classes); classes outside the corpus are best-effort.
+- Lone \uD800–\uDFFF surrogates: serde Values cannot carry them (Python builds
+  the char); excluded from the corpus.
+- CLI crash tracebacks (exit 1) surface as `__CRASH__:<class envelope>` on
+  stdout — the machine-path-dependent traceback text stays out of the corpus.