Merge pull request #646 from escoffier-labs/fix/636-checkpoint-export-privacy

fix(runs): strip private recovery-checkpoint bodies at export (#636)

Author: solomonneas

src/brigade/run_checkpoint.py

@@ -1178,3 +1178,98 @@ def recover_from_checkpoint(
         except run_projector.ProjectionError as exc:
             raise CheckpointError(_bound("projection failed"), category="projection") from exc
     return _restore_run_json_from_checkpoint(run_dir, restore_bytes, run_meta=run_meta)
+
+
+# -- Export boundary (issue #636) ---------------------------------------------
+
+_ARTIFACT_REFERENCE_KEYS = frozenset({"path", "sha256", "media_type", "byte_size", "privacy_class"})
+
+
+def checkpoint_artifact_reference(*, sha256: str, byte_size: int) -> dict[str, Any]:
+    """Return the closed artifact-reference shape used at export boundaries."""
+    if not isinstance(sha256, str) or not _HEX64.fullmatch(sha256):
+        raise CheckpointError(_bound("checkpoint export digest is invalid"), category="export-privacy")
+    if isinstance(byte_size, bool) or not isinstance(byte_size, int) or byte_size < 0:
+        raise CheckpointError(_bound("checkpoint export byte size is invalid"), category="export-privacy")
+    return {
+        "path": f"events/{CHECKPOINT_DIR_NAME}/{sha256}.json",
+        "sha256": sha256,
+        "media_type": CHECKPOINT_MEDIA_TYPE,
+        "byte_size": byte_size,
+        "privacy_class": CHECKPOINT_PRIVACY_CLASS,
+    }
+
+
+def refuse_checkpoint_body_export(*, reason: str = "checkpoint body is private") -> None:
+    """Refuse an export that would emit checkpoint body content."""
+    raise CheckpointError(
+        _bound(f"{reason} (privacy_class={CHECKPOINT_PRIVACY_CLASS})"),
+        category="export-privacy",
+    )
+
+
+def is_checkpoint_artifact_reference(payload: Mapping[str, Any]) -> bool:
+    """True when payload is exactly the closed artifact-reference shape."""
+    if set(payload) != _ARTIFACT_REFERENCE_KEYS:
+        return False
+    sha = payload.get("sha256")
+    path = payload.get("path")
+    byte_size = payload.get("byte_size")
+    return (
+        isinstance(sha, str)
+        and bool(_HEX64.fullmatch(sha))
+        and path == f"events/{CHECKPOINT_DIR_NAME}/{sha}.json"
+        and payload.get("media_type") == CHECKPOINT_MEDIA_TYPE
+        and payload.get("privacy_class") == CHECKPOINT_PRIVACY_CLASS
+        and not isinstance(byte_size, bool)
+        and isinstance(byte_size, int)
+        and byte_size >= 0
+    )
+
+
+def strip_checkpoint_bodies_for_export(run_dir: Path) -> list[dict[str, Any]]:
+    """Replace recovery-checkpoint file bodies with artifact references.
+
+    Local recovery continues to read the original private bodies under the run
+    directory. Call this only on a copy that is about to leave the run tree
+    (archive, bundle, sync, or similar export boundary).
+    """
+    cp_dir = checkpoint_dir(run_dir)
+    if not cp_dir.is_dir():
+        return []
+    replaced: list[dict[str, Any]] = []
+    for path in sorted(cp_dir.glob("*.json")):
+        if not path.is_file() or path.is_symlink():
+            refuse_checkpoint_body_export(reason="checkpoint export path is not a regular file")
+        raw = path.read_bytes()
+        try:
+            parsed = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            parsed = None
+        if isinstance(parsed, dict) and is_checkpoint_artifact_reference(parsed):
+            replaced.append(dict(parsed))
+            continue
+        sha = hashlib.sha256(raw).hexdigest()
+        if path.name != f"{sha}.json":
+            refuse_checkpoint_body_export(reason="checkpoint export filename digest mismatch")
+        reference = checkpoint_artifact_reference(sha256=sha, byte_size=len(raw))
+        path.write_text(json.dumps(reference, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        os.chmod(path, 0o600)
+        replaced.append(reference)
+    return replaced
+
+
+def assert_export_tree_has_no_checkpoint_bodies(root: Path) -> None:
+    """Fail closed if any recovery-checkpoint file under root still holds a body."""
+    for path in sorted(Path(root).rglob(f"*/{CHECKPOINT_DIR_NAME}/*.json")):
+        if not path.is_file():
+            continue
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise CheckpointError(
+                _bound("checkpoint export body is unreadable"),
+                category="export-privacy",
+            ) from exc
+        if not isinstance(payload, dict) or not is_checkpoint_artifact_reference(payload):
+            refuse_checkpoint_body_export(reason="checkpoint body crossed an export boundary")

src/brigade/run_events.py

@@ -144,6 +144,16 @@
     }
 )
 
+# Recovery-checkpoint bodies under ``events/recovery-checkpoints/`` are
+# ``privacy_class: private`` (see ``run_checkpoint.CHECKPOINT_PRIVACY_CLASS``).
+# Journal event payloads only carry the closed artifact-reference shape
+# (relative path, sha256, media type, byte size, privacy class). Any exporter
+# or collector that copies a run directory across a boundary must strip those
+# bodies, replace them with that artifact-reference shape, or refuse with a
+# bounded error naming the privacy class. Local recovery may still read the
+# private bodies in place. Helpers: ``run_checkpoint.strip_checkpoint_bodies_for_export``,
+# ``run_checkpoint.refuse_checkpoint_body_export``.
+
 
 class CanonicalizationError(ValueError):
     """Raised when a value cannot be canonicalized under the strict rules."""

src/brigade/work_cmd/verification.py

@@ -551,31 +551,68 @@ def _verify_archive_tree_manifest(root: Path) -> dict[str, tuple[str, str | None
     return manifest
 
 
+def _expected_verify_archive_manifest(
+    run_dir: Path, source_manifest: dict[str, tuple[str, str | None]]
+) -> dict[str, tuple[str, str | None]]:
+    """Return the tree manifest after recovery-checkpoint export stripping (#636)."""
+    from brigade import run_checkpoint
+
+    expected = dict(source_manifest)
+    prefix = f"events/{run_checkpoint.CHECKPOINT_DIR_NAME}/"
+    for relative, (kind, _digest) in list(expected.items()):
+        if kind != "file" or not relative.startswith(prefix) or not relative.endswith(".json"):
+            continue
+        path = run_dir / relative
+        raw = path.read_bytes()
+        try:
+            parsed = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            parsed = None
+        if isinstance(parsed, dict) and run_checkpoint.is_checkpoint_artifact_reference(parsed):
+            continue
+        sha = hashlib.sha256(raw).hexdigest()
+        if path.name != f"{sha}.json":
+            raise OSError(f"verify archive checkpoint export filename digest mismatch: {relative}")
+        reference = run_checkpoint.checkpoint_artifact_reference(sha256=sha, byte_size=len(raw))
+        ref_bytes = (json.dumps(reference, indent=2, sort_keys=True) + "\n").encode("utf-8")
+        expected[relative] = ("file", hashlib.sha256(ref_bytes).hexdigest())
+    return expected
+
+
 def _archive_verify_run(run_dir: Path, archive_root: Path) -> dict[str, Any]:
     """Copy one run dir into the archive and append an index entry. Raises on failure.
 
     Integrity is checked twice: the archived receipt bytes must match the source
     bytes, and a receipt that carries ``digests.receipt_sha256`` must still
     re-hash to that value after the copy. Callers must treat any exception as
-    "do not delete the original".
+    "do not delete the original". Recovery-checkpoint bodies are private and are
+    replaced with the closed artifact-reference shape before the archive lands
+    (issue #636); local source run dirs are left unchanged.
     """
+    from brigade import run_checkpoint
+
     run_id = run_dir.name
     dest = archive_root / run_id
     source_manifest = _verify_archive_tree_manifest(run_dir)
+    expected_manifest = _expected_verify_archive_manifest(run_dir, source_manifest)
     receipt_path = run_dir / "receipt.json"
     source_receipt_sha = localio.file_sha256(receipt_path) if receipt_path.is_file() else None
     if dest.is_symlink():
         raise OSError(f"verify archive conflict: {dest} is a symlink")
     if dest.exists():
-        # Re-archive of the same run id is only safe when the evidence is identical.
-        if _verify_archive_tree_manifest(dest) != source_manifest:
+        # Re-archive is safe when evidence matches the source, or when the
+        # existing archive is already the privacy-normalized export of source.
+        dest_manifest = _verify_archive_tree_manifest(dest)
+        if dest_manifest not in (source_manifest, expected_manifest):
             raise OSError(f"verify archive conflict: {dest} already exists with different evidence")
         dest_receipt = dest / "receipt.json"
         dest_sha = localio.file_sha256(dest_receipt) if dest_receipt.is_file() else None
         if dest_sha != source_receipt_sha:
             raise OSError(f"verify archive conflict: {dest} already exists with different evidence")
         if dest_sha is not None:
             _assert_archived_receipt_integrity(dest_receipt)
+        run_checkpoint.strip_checkpoint_bodies_for_export(dest)
+        run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(dest)
         entry = _verify_archive_index_entry(run_dir, dest, dest_sha, already_archived=True)
         _append_verify_archive_index(archive_root, entry)
         return entry
@@ -590,6 +627,13 @@ def _archive_verify_run(run_dir: Path, archive_root: Path) -> dict[str, Any]:
             if copied_sha != source_receipt_sha:
                 raise OSError(f"verify archive copy integrity check failed: {run_id}")
             _assert_archived_receipt_integrity(staging / "receipt.json")
+        # Export boundary (#636): never archive private recovery-checkpoint
+        # bodies. Replace them with the closed artifact-reference shape after
+        # the byte-identical copy check so local recovery semantics stay intact.
+        run_checkpoint.strip_checkpoint_bodies_for_export(staging)
+        run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(staging)
+        if _verify_archive_tree_manifest(staging) != expected_manifest:
+            raise OSError(f"verify archive export privacy normalize mismatch: {run_id}")
         os.rename(staging, dest)
         if _verify_archive_tree_manifest(run_dir) != source_manifest:
             raise OSError(f"verify archive source changed during copy: {run_id}")

tests/test_run_checkpoint.py

@@ -9,6 +9,7 @@
 import hashlib
 import json
 import os
+import shutil
 import signal
 import stat
 import subprocess
@@ -3266,3 +3267,80 @@ def test_write_text_atomic_sigkill_during_temp_window_preserves_one_valid_payloa
     )
     for run_dir in attempt_run_dirs:
         assert _write_text_atomic_temp_paths(run_dir) == []
+
+
+# -- Issue #636: checkpoint export privacy -----------------------------------
+
+
+def test_strip_checkpoint_bodies_for_export_replaces_private_body(tmp_path):
+    run_dir = _run_dir(tmp_path)
+    private_task = "SECRET_TASK_PROMPT_do_not_export"
+    private_error = "SECRET_ERROR_TRACE_do_not_export"
+    body = _writer_bytes(
+        {
+            "schema": "brigade.run.v1",
+            "status": "failed",
+            "task": private_task,
+            "error": private_error,
+        }
+    )
+    placed = _place_checkpoint_file(run_dir, body)
+    assert private_task in placed.read_text(encoding="utf-8")
+
+    export_copy = tmp_path / "export-copy"
+    shutil.copytree(run_dir, export_copy)
+
+    refs = run_checkpoint.strip_checkpoint_bodies_for_export(export_copy)
+    assert len(refs) == 1
+    assert run_checkpoint.is_checkpoint_artifact_reference(refs[0])
+    assert refs[0]["privacy_class"] == "private"
+    assert refs[0]["sha256"] == hashlib.sha256(body).hexdigest()
+    assert refs[0]["byte_size"] == len(body)
+
+    exported = (export_copy / "events" / "recovery-checkpoints" / placed.name).read_text(encoding="utf-8")
+    assert private_task not in exported
+    assert private_error not in exported
+    assert '"task"' not in exported
+    assert json.loads(exported) == refs[0]
+
+    # Local recovery source is unchanged.
+    assert placed.read_bytes() == body
+    run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(export_copy)
+
+
+def test_assert_export_tree_refuses_checkpoint_body(tmp_path):
+    run_dir = _run_dir(tmp_path)
+    body = _writer_bytes({"schema": "brigade.run.v1", "status": "planning", "task": "keep-private"})
+    _place_checkpoint_file(run_dir, body)
+
+    with pytest.raises(run_checkpoint.CheckpointError, match="privacy_class=private") as exc_info:
+        run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(run_dir)
+    assert exc_info.value.category == "export-privacy"
+
+
+def test_refuse_checkpoint_body_export_names_privacy_class():
+    with pytest.raises(run_checkpoint.CheckpointError, match="privacy_class=private") as exc_info:
+        run_checkpoint.refuse_checkpoint_body_export()
+    assert exc_info.value.category == "export-privacy"
+
+
+def test_local_recovery_round_trips_after_export_strip_of_copy(tmp_path):
+    """Export stripping a copy must not change local recovery semantics."""
+    workspace = _workspace(tmp_path)
+    run_dir = _run_dir(tmp_path)
+    run_json_obj = {
+        "schema": "brigade.run.v1",
+        "status": "planning",
+        "task": "SECRET_TASK_for_local_recovery_only",
+    }
+    _activated_journal_with_checkpoint(workspace, run_dir, run_json_obj)
+
+    export_copy = tmp_path / "export-copy"
+    shutil.copytree(run_dir, export_copy)
+    run_checkpoint.strip_checkpoint_bodies_for_export(export_copy)
+    run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(export_copy)
+
+    (run_dir / "run.json").unlink()
+    repaired = run_checkpoint.recover_from_checkpoint(run_dir, None)
+    assert repaired == run_json_obj
+    assert (run_dir / "run.json").read_bytes() == _writer_bytes(run_json_obj)

tests/test_work_cmd_verification.py

@@ -2668,3 +2668,36 @@ def test_verify_receipt_identity_preserves_real_index(tmp_target, monkeypatch, c
         text=True,
     ).stdout
     assert after == before
+
+
+def test_archive_verify_run_strips_recovery_checkpoint_bodies(tmp_path):
+    """Export boundary: verify-archive must not emit private checkpoint bodies (#636)."""
+    from brigade import run_checkpoint
+    from brigade.work_cmd import helpers, verification
+
+    root = helpers._verify_runs_root(tmp_path)
+    root.mkdir(parents=True)
+    run_dir, _ = _write_verify_run_dir(root, "20260101-000001-a", sign=True)
+    _write_verify_run_dir(root, "20260101-000002-b")
+
+    private_task = "SECRET_CHECKPOINT_TASK_must_not_archive"
+    body_obj = {"schema": "brigade.run.v1", "status": "failed", "task": private_task, "error": "boom"}
+    body = (json.dumps(body_obj, indent=2, sort_keys=True) + "\n").encode("utf-8")
+    cp_dir = run_dir / "events" / "recovery-checkpoints"
+    cp_dir.mkdir(parents=True)
+    sha = hashlib.sha256(body).hexdigest()
+    (cp_dir / f"{sha}.json").write_bytes(body)
+    os.chmod(cp_dir / f"{sha}.json", 0o600)
+
+    archive_root = tmp_path / "verify-archive"
+    removed = verification._prune_verify_runs(tmp_path, keep=1, archive_root=archive_root)
+
+    assert removed == 1
+    archived_cp = archive_root / "20260101-000001-a" / "events" / "recovery-checkpoints" / f"{sha}.json"
+    assert archived_cp.is_file()
+    payload = json.loads(archived_cp.read_text(encoding="utf-8"))
+    assert run_checkpoint.is_checkpoint_artifact_reference(payload)
+    assert private_task not in archived_cp.read_text(encoding="utf-8")
+    assert '"task"' not in archived_cp.read_text(encoding="utf-8")
+    # Source verify-run is deleted after archive; the privacy rule is on the export.
+    assert not (root / "20260101-000001-a").exists()