feat(delegate): attest patch apply/reject decisions into the acceptance packet (D-trace)

Phase O-F of the harness-health sprint (owner decision 4=A):

integrate_delegated_patch applied a child's diff on five mechanical
manifest fields with zero review facts anywhere on the path — and none
exist to carry: the child is forbidden from reviewing its own diff and no
host review of the captured bytes runs. Rather than dressing the child's
own claim up as a review (a provenance lie) or gating the apply (a
convergence tax the owner rejected), the APPLY itself is attested:

- _write_verdict additionally lands a typed delegate_run_patch_verdict
  custody row (family-prefixed — an unprefixed spelling is invisible to
  _iter_rows, the known dead-letter class) carrying disposition, applied,
  bounded reason, patch sha and the pipeline (delegated run_<rid> subjects
  vs subagent children, classified at the one writer). A failed verdict-
  artifact write is disclosed on the row instead of dying as a silent "".
- delegate_evidence.acceptance_patch_dispositions projects those rows as a
  bounded host-attested acceptance-packet section (capability_deltas
  shape: cap 20 + exact omitted count, newest kept) with the honest
  headline unreviewed_delegated_apply when a delegated patch landed.
  Absence = no disposition recorded, never "reviewed clean"; an
  unreadable log is the typed evidence_read_failed marker.
- review_evidence wires the section beside capability_deltas/substrate.

No gate on apply anywhere. The nanny-wake copy was considered and cut:
at wake time no disposition exists yet, and a sometimes-present field
recreates the absence-ambiguity this section exists to kill.

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>

ouroboros/delegate_evidence.py

@@ -301,6 +301,65 @@ def acceptance_substrate_facts(ctx: Any, task_id: str) -> Dict[str, Any]:
 
 
 _ACCEPT_DELTA_CHILD_CAP = 20  # reduced-children rows in the finalizer aggregate
+_ACCEPT_PATCH_DISPOSITION_CAP = 20  # disposition rows in the acceptance section
+
+
+def acceptance_patch_dispositions(drive_root: Any, task_id: str) -> Dict[str, Any]:
+    """Typed aggregate of this parent's patch apply/reject decisions (D-trace).
+
+    ``integrate_delegated_patch`` applied a child's diff on FIVE mechanical
+    manifest fields with zero review facts anywhere on the path; the owner
+    decision (4=A) is to ATTEST the apply rather than invent a review: every
+    verdict now also lands as a ``subagent_patch_verdict`` custody row, and
+    this section projects those rows host-attested into the acceptance packet
+    (the ``capability_deltas`` shape — bounded, first-class, never squeezed
+    through the 4KB artifact-preview cliff). ABSENCE of the section means "no
+    disposition recorded", never "reviewed clean"; an unreadable log is the
+    typed ``evidence_read_failed`` marker, never an empty-therefore-clean
+    section (the ``task_execution_evidence`` rule, GR6-4).
+    """
+    from ouroboros import delegate_custody as custody
+    from ouroboros.utils import truncate_review_artifact
+
+    tid = str(task_id or "")
+    out: Dict[str, Any] = {}
+    log_path = custody.event_log_path(drive_root)
+    try:
+        if log_path.exists():
+            with log_path.open("rb"):
+                pass
+        else:
+            return out
+    except OSError:
+        return {"evidence_read_failed": True}
+    rows: List[Dict[str, Any]] = []
+    for row in custody._iter_rows(log_path):
+        if str(row.get("type") or "") != "delegate_run_patch_verdict":
+            continue
+        if str(row.get("task_id") or "") != tid:
+            continue
+        rows.append({
+            "child": str(row.get("child_task_id") or ""),
+            "pipeline": str(row.get("pipeline") or ""),
+            "disposition": str(row.get("disposition") or ""),
+            "applied": bool(row.get("applied")),
+            "reason": truncate_review_artifact(str(row.get("reason") or ""), limit=600),
+            "patch_sha256": str(row.get("patch_sha256") or ""),
+            **({"verdict_artifact_write_failed": True}
+               if row.get("verdict_artifact_write_failed") else {}),
+        })
+    if not rows:
+        return out
+    out["total"] = len(rows)
+    if len(rows) > _ACCEPT_PATCH_DISPOSITION_CAP:
+        out["omitted"] = len(rows) - _ACCEPT_PATCH_DISPOSITION_CAP
+        rows = rows[-_ACCEPT_PATCH_DISPOSITION_CAP:]
+    out["rows"] = rows
+    if any(r["applied"] and r.get("pipeline") == "delegated" for r in rows):
+        # The honest headline the panel weighs: a delegated patch landed with
+        # no host-side review of its bytes (there is none on this path).
+        out["unreviewed_delegated_apply"] = True
+    return out
 
 
 def acceptance_capability_deltas(drive_root: Any, task_id: str, root_task_id: str) -> Dict[str, Any]:

ouroboros/review_evidence.py

@@ -883,6 +883,7 @@ def build_task_acceptance_evidence(
             prov["mutation_attribution"] = "host_attested"
         from ouroboros.delegate_evidence import (
             acceptance_capability_deltas,
+            acceptance_patch_dispositions,
             acceptance_substrate_facts,
         )
 
@@ -892,6 +893,12 @@ def build_task_acceptance_evidence(
         if substrate_facts := acceptance_substrate_facts(ctx, task_id):
             ev["substrate_execution"] = redact_projection(substrate_facts).value
             prov["substrate_execution"] = "host_attested"
+        # D-trace (owner 4=A): the parent's patch apply/reject attestations —
+        # visibility for the panel, never a gate on apply. Absence = no
+        # disposition recorded, not "reviewed clean".
+        if patch_dispositions := acceptance_patch_dispositions(drive_root, task_id):
+            ev["delegated_patch_dispositions"] = redact_projection(patch_dispositions).value
+            prov["delegated_patch_dispositions"] = "host_attested"
     repo_diff = collect_turn_diff(ctx, include_recent_commit=include_recent_commit)
     diff_meta: Dict[str, Any] = {}
     if "OMISSION NOTE: truncated at " in str(repo_diff or "") or "... (truncated from " in str(repo_diff or ""):

ouroboros/tools/subagent_integration.py

@@ -160,11 +160,38 @@ def _write_verdict(
         "diffstat": str((manifest or {}).get("diffstat") or ""),
     }
     path = art_dir / f"subagent_patch_verdict_{child_task_id}.json"
+    artifact_write_failed = False
     try:
         atomic_write_json(path, verdict, trailing_newline=True)
     except Exception:
-        return ""
-    return str(path)
+        artifact_write_failed = True
+    # D-trace: the apply/reject decision also lands as a typed row in the
+    # custody event log, so the acceptance packet builds the disposition
+    # section from ONE replayable store for both patch pipelines (the
+    # delegated pipeline's run-keyed PATCH_DISPOSED row rides beside it; the
+    # scanner splits by `tool`, so nothing double-counts). A failed artifact
+    # write is disclosed on the row rather than dying as a silent "".
+    try:
+        from ouroboros import delegate_custody as custody
+
+        custody.emit(getattr(ctx, "drive_root", "."), "delegate_run_patch_verdict", {
+            "run_id": "",
+            "task_id": parent_task_id,
+            "child_task_id": child_task_id,
+            # The delegated pipeline mints its verdict subject as run_<rid>
+            # (every _integrate_delegated_patch call site); classifying at
+            # the ONE writer keeps that convention local instead of leaking
+            # prefix-matching into readers.
+            "pipeline": "delegated" if child_task_id.startswith("run_") else "subagent",
+            "disposition": outcome,
+            "applied": bool(applied),
+            "reason": str(reason or "")[:600],
+            "patch_sha256": str(verdict.get("patch_sha256") or ""),
+            "verdict_artifact_write_failed": artifact_write_failed,
+        })
+    except Exception:
+        log.debug("subagent patch verdict custody row failed", exc_info=True)
+    return "" if artifact_write_failed else str(path)
 
 
 def _child_write_root(child_result: Dict[str, Any]) -> str:

tests/test_patch_disposition_trace.py

@@ -0,0 +1,109 @@
+"""D-trace (owner 4=A): patch apply/reject decisions are attested, not gated.
+
+``integrate_delegated_patch`` applies a child's diff on five mechanical
+manifest fields; no review facts exist anywhere on the path (the child is
+forbidden from reviewing its own diff, and no host review of the captured
+bytes runs). Rather than inventing a review, every verdict lands as a typed
+``subagent_patch_verdict`` custody row and the acceptance packet carries the
+host-attested disposition section — first-class, bounded, never squeezed
+through the 4KB artifact-preview cliff. Nothing refuses an apply.
+"""
+
+import json
+
+from ouroboros import delegate_custody as custody
+from ouroboros.delegate_evidence import acceptance_patch_dispositions
+
+
+def _emit_verdict(tmp_path, *, task_id="t-parent", child="run_r1",
+                  pipeline="delegated", disposition="applied", applied=True,
+                  reason="looks right", sha="abc123", write_failed=False):
+    assert custody.emit(tmp_path, "delegate_run_patch_verdict", {
+        "run_id": "", "task_id": task_id, "child_task_id": child,
+        "pipeline": pipeline, "disposition": disposition, "applied": applied,
+        "reason": reason, "patch_sha256": sha,
+        "verdict_artifact_write_failed": write_failed,
+    })
+
+
+def test_no_rows_means_no_section_not_clean(tmp_path):
+    assert acceptance_patch_dispositions(tmp_path, "t-parent") == {}
+
+
+def test_dispositions_project_with_the_unreviewed_headline(tmp_path):
+    _emit_verdict(tmp_path)
+    _emit_verdict(tmp_path, child="t-sub", pipeline="subagent",
+                  disposition="rejected", applied=False, reason="conflicts")
+    section = acceptance_patch_dispositions(tmp_path, "t-parent")
+    assert section["total"] == 2
+    rows = {r["child"]: r for r in section["rows"]}
+    assert rows["run_r1"]["applied"] is True
+    assert rows["run_r1"]["pipeline"] == "delegated"
+    assert rows["t-sub"]["applied"] is False
+    # The honest headline: a delegated patch landed with no host review of
+    # its bytes. Visibility only — nothing on the apply path refuses.
+    assert section["unreviewed_delegated_apply"] is True
+
+
+def test_rejected_only_delegated_rows_carry_no_apply_headline(tmp_path):
+    _emit_verdict(tmp_path, disposition="rejected", applied=False)
+    section = acceptance_patch_dispositions(tmp_path, "t-parent")
+    assert "unreviewed_delegated_apply" not in section
+
+
+def test_other_tasks_rows_stay_out(tmp_path):
+    _emit_verdict(tmp_path, task_id="t-other")
+    assert acceptance_patch_dispositions(tmp_path, "t-parent") == {}
+
+
+def test_bounded_with_exact_omitted_count(tmp_path):
+    for i in range(25):
+        _emit_verdict(tmp_path, child=f"run_r{i}")
+    section = acceptance_patch_dispositions(tmp_path, "t-parent")
+    assert section["total"] == 25
+    assert section["omitted"] == 5
+    assert len(section["rows"]) == 20
+    # Newest rows survive the bound (the most recent decisions matter most).
+    assert section["rows"][-1]["child"] == "run_r24"
+
+
+def test_write_verdict_emits_the_custody_row(tmp_path):
+    from types import SimpleNamespace
+
+    from ouroboros.tools.subagent_integration import _write_verdict
+
+    ctx = SimpleNamespace(drive_root=str(tmp_path), task_id="t-parent",
+                          task_metadata={}, budget_drive_root=str(tmp_path))
+    path = _write_verdict(
+        ctx, "run_r9", outcome="applied", reason="ok", files=["a.py"],
+        manifest={"sha256": "deadbeef", "diffstat": "1 file"},
+        applied=True, conflicts=[], protected=[],
+    )
+    assert path, "verdict artifact write should succeed in a temp drive"
+    rows = [
+        row for row in custody._iter_rows(custody.event_log_path(tmp_path))
+        if row.get("type") == "delegate_run_patch_verdict"
+    ]
+    assert len(rows) == 1
+    row = rows[0]
+    assert row["child_task_id"] == "run_r9"
+    assert row["pipeline"] == "delegated"
+    assert row["patch_sha256"] == "deadbeef"
+    assert row["applied"] is True
+    assert row["verdict_artifact_write_failed"] is False
+
+
+def test_verdict_artifact_content_still_written(tmp_path):
+    from types import SimpleNamespace
+
+    from ouroboros.tools.subagent_integration import _write_verdict
+
+    ctx = SimpleNamespace(drive_root=str(tmp_path), task_id="t-parent",
+                          task_metadata={}, budget_drive_root=str(tmp_path))
+    path = _write_verdict(
+        ctx, "child-7", outcome="rejected", reason="conflicts", files=[],
+        manifest={}, applied=False, conflicts=["x"], protected=[],
+    )
+    data = json.loads(open(path).read())
+    assert data["outcome"] == "rejected"
+    assert data["child_task_id"] == "child-7"