Persist execution binding evidence and tighten capture reject

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

ouroboros/delegate_custody.py

@@ -171,6 +171,7 @@ class RunCustody:
     # where to diff, startup GC tells live snapshots from disposable ones.
     snapshot_id: str = ""
     execution_root: str = ""
+    execution_binding_fingerprint: str = ""
     baseline_sha: str = ""
     target_root: str = ""
     authority_source: str = ""
@@ -760,6 +761,7 @@ def invocation_record(drive_root: Any, invocation_id: str, *,
                 # target the original attempt bound — never a re-derivation.
                 "snapshot_id": str(row.get("snapshot_id") or ""),
                 "execution_root": str(row.get("execution_root") or ""),
+                "execution_binding_fingerprint": str(row.get("execution_binding_fingerprint") or ""),
                 "baseline_sha": str(row.get("baseline_sha") or ""),
                 "target_root": str(row.get("target_root") or ""),
                 "authority_source": str(row.get("authority_source") or ""),

ouroboros/delegate_registration_policy.py

@@ -90,7 +90,7 @@ STARTED_STR_FIELDS: Tuple[Tuple[str, str], ...] = tuple(
         "task_id", "route_id", "model", "profile_id", "project_id", "root_task_id",
         "parent_task_id", "category", "source", *REVIEW_ATTRIBUTION_KEYS,
         "ledger_root", "idempotency_key", "invocation_id",
-        "snapshot_id", "execution_root", "baseline_sha", "target_root",
+        "snapshot_id", "execution_root", "execution_binding_fingerprint", "baseline_sha", "target_root",
         "authority_source", "access", "mode", "isolation",
         "selected_subagent_id", "config_fingerprint", "work_order_fingerprint",
         "work_order_coverage", "authority_fingerprint",
@@ -109,7 +109,7 @@ STARTED_PROGRESS_FLAGS: Tuple[str, ...] = (
 # may be minted by a context that no longer knows the original binding; the
 # first recorded fact is authoritative and is never erased or retargeted.
 STARTED_FIRST_WINS_FACTS: Tuple[str, ...] = (
-    "snapshot_id", "execution_root", "baseline_sha", "target_root",
+    "snapshot_id", "execution_root", "execution_binding_fingerprint", "baseline_sha", "target_root",
     "authority_source", "resource_ref", "selected_subagent_id",
     "config_fingerprint", "work_order_fingerprint", "work_order_coverage",
     "authority_fingerprint", "work_order_source_request", "category", "source",

ouroboros/delegate_source_coverage.py

@@ -211,7 +211,8 @@ def record_started_custody(
     selected_subagent_id: str,
     config_fingerprint: str, work_order_fingerprint: str, work_order_coverage: str,
     work_order_source_request: Dict[str, Any], authority_fingerprint: str,
-    snapshot_id: str, target_root: str, baseline_sha: str, authority_source: str,
+    snapshot_id: str, execution_binding_fingerprint: str, target_root: str,
+    baseline_sha: str, authority_source: str,
     resource_ref: Dict[str, Any], capture_mode: str, processing: Mapping[str, Any] | None = None,
 ) -> bool:
     """Write the one STARTED custody row, including the source binding."""
@@ -245,6 +246,7 @@ def record_started_custody(
         snapshot_id=snapshot_id,
         execution_root=(root if snapshot_id or (resource_ref.get("workspace_kind") == "directory"
                                                and resource_ref.get("strategy") == "direct") else ""),
+        execution_binding_fingerprint=execution_binding_fingerprint,
         baseline_sha=baseline_sha,
         target_root=target_root,
         authority_source=authority_source,

ouroboros/delegate_start_instructions.py

@@ -83,12 +83,29 @@ def execution_binding_instruction(execution_root: str, authority_root: str) -> s
     omission or by appearing earlier in the work order.
     """
     execution = str(execution_root or "").strip()
+    if not execution:
+        return ""
+    return execution_binding_text(execution_root, authority_root)
+
+
+def execution_binding_fingerprint(execution_root: str, authority_root: str,
+                                  binding_kind: str = "snapshot") -> str:
+    """Digest the typed execution/authority binding stored with custody."""
+    execution = str(execution_root or "").strip()
+    authority = str(authority_root or "").strip()
+    return sha256(f"{binding_kind}\0{execution}\0{authority}".encode("utf-8")).hexdigest()
+
+
+def execution_binding_text(execution_root: str, authority_root: str,
+                           binding_sha256: str = "") -> str:
+    execution = str(execution_root or "").strip()
     authority = str(authority_root or "").strip()
     if not execution:
         return ""
+    digest = binding_sha256 or execution_binding_fingerprint(execution, authority)
     return (
         "\n\nDELEGATED EXECUTION BINDING (separate typed host fact; the canonical "
-        "work order remains byte-identical): "
+        f"work order remains byte-identical; sha256={digest}): "
         f"the sole writable execution root for this run is {execution}. "
         "Use relative paths or absolute paths under that root for every shell, "
         "file, and patch operation. The stable authority/project root "
@@ -100,12 +117,13 @@ def execution_binding_instruction(execution_root: str, authority_root: str) -> s
     )
 
 
-def directory_copy_binding_instruction(authority_root: str) -> str:
+def directory_copy_binding_instruction(authority_root: str, binding_sha256: str = "") -> str:
     """Tell a directory-copy child that the engine creates its write root later."""
     authority = str(authority_root or "").strip() or "(unknown)"
+    digest = binding_sha256 or execution_binding_fingerprint("", authority, "directory_copy")
     return (
         "\n\nDELEGATED DIRECTORY COPY BINDING (separate typed host fact; the canonical "
-        "work order remains byte-identical): the engine will create a private "
+        f"work order remains byte-identical; sha256={digest}): the engine will create a private "
         "execution copy for this run. Use only the engine-provided working "
         "directory/cwd for writes; the selected authority folder "
         f"{authority} is a read-only source reference. Do not write to that "
@@ -114,9 +132,9 @@ def directory_copy_binding_instruction(authority_root: str) -> str:
 
 
 def apply_execution_binding(instructions: str, execution_root: str,
-                            authority_root: str) -> str:
+                            authority_root: str, binding_sha256: str = "") -> str:
     """Append one typed binding without rewriting canonical work-order bytes."""
-    return instructions + execution_binding_instruction(execution_root, authority_root)
+    return instructions + execution_binding_text(execution_root, authority_root, binding_sha256)
 
 
 def append_coordination_context(

ouroboros/delegate_target_drift.py

@@ -116,6 +116,7 @@ def _persist_target_drift(manifest_path: pathlib.Path, manifest: Dict[str, Any],
     from ouroboros.utils import atomic_write_json, utc_now_iso
 
     updated = dict(manifest)
+    updated["authority_drift_source_status"] = str(manifest.get("status") or "")
     updated["authority_drift"] = {
         "checked": bool(evidence.get("checked")),
         "paths": list(evidence.get("paths") or []),

ouroboros/tools/delegate.py

@@ -58,6 +58,7 @@ from ouroboros.delegate_start_instructions import (
     append_coordination_context,
     apply_execution_binding,
     directory_copy_binding_instruction,
+    execution_binding_fingerprint,
 )
 from ouroboros.subagent_runtime import (  # noqa: F401 - shared primitive re-export
     delegate_start_entry as _delegate_start_entry,
@@ -317,6 +318,17 @@ def _processing_start_request(request, actor, gateway, route):
                      "reason": "submitted" if "processingPreference" in request else "processing_not_submitted"}
 
 
+def _start_binding_fingerprint(current: str, root: str, target_root: str,
+                               snapshot_id: str, resource_ref: Dict[str, Any]) -> str:
+    if current:
+        return current
+    if snapshot_id:
+        return execution_binding_fingerprint(root, target_root, "snapshot")
+    if resource_ref.get("strategy") == "copy":
+        return execution_binding_fingerprint("", target_root, "directory_copy")
+    return ""
+
+
 def _delegate_start(ctx: ToolContext, prompt: str, max_seconds: Optional[int] = None,
                     retry_of: Optional[str] = None, root: Optional[str] = None,
                     bucket: Optional[str] = None, skill_name: Optional[str] = None,
@@ -349,6 +361,7 @@ def _delegate_start(ctx: ToolContext, prompt: str, max_seconds: Optional[int] =
     drive = custody.custody_root(ctx)
     owned_project_id, project_persistent = "", False
     invocation_id = snapshot_id = baseline_sha = target_root = authority_source = ""
+    binding_fingerprint = ""
     processing_info: Dict[str, Any] = {}
     resource_ref, directory_options = {}, {}
     retry_token = str(retry_of or "").strip()
@@ -473,10 +486,13 @@ def _delegate_start(ctx: ToolContext, prompt: str, max_seconds: Optional[int] =
             execution_root = (root if directory_options.get("isolation") == "live" else "") if directory_options else delegated_execution_workspace_root(gateway, authority, root)
             scope_root = target_root if execution_root or directory_options else root
             if snapshot is not None:
+                binding_fingerprint = execution_binding_fingerprint(
+                    execution_root or root, target_root)
                 instructions = apply_execution_binding(
-                    instructions, execution_root or root, target_root)
+                    instructions, execution_root or root, target_root, binding_fingerprint)
             elif directory_options and directory_options.get("isolation") == "envelope":
-                instructions += directory_copy_binding_instruction(target_root)
+                binding_fingerprint = execution_binding_fingerprint("", target_root, "directory_copy")
+                instructions += directory_copy_binding_instruction(target_root, binding_fingerprint)
             (project_id, owned_project_id, project_persistent) = resolve_registration(
                 gateway, scope_root, execution_root, getattr(authority, "access", ""))
             if directory_options:
@@ -503,6 +519,8 @@ def _delegate_start(ctx: ToolContext, prompt: str, max_seconds: Optional[int] =
             project_owned=bool(owned_project_id), project_persistent=project_persistent, route=route.route_id,
             root_task_id=str(lineage.get("root_task_id") or ""), parent_task_id=str(lineage.get("parent_task_id") or ""),
             snapshot_id=snapshot_id, execution_root=(root if snapshot_id or resource_ref.get("strategy") == "direct" else ""),
+            execution_binding_fingerprint=_start_binding_fingerprint(
+                binding_fingerprint, root, target_root, snapshot_id, resource_ref),
             baseline_sha=baseline_sha, target_root=target_root,
             authority_source=authority_source, resource_ref=resource_ref,
             # Recovery proves the original actor and compiled brief before adoption.
@@ -525,11 +543,6 @@ def _delegate_start(ctx: ToolContext, prompt: str, max_seconds: Optional[int] =
                 ),
             )
         if not requested:
-            # The POST is CONDITIONAL on the durable request row: a run started
-            # without it is live and unfindable if this worker dies. A fresh
-            # start's registration is definitively retirable; a RETRY's project
-            # belongs to the original attempt, whose POST may have bound a live
-            # run — its fate stays unknown and its invocation stays pending.
             return _fail(
                 "delegate_start", "start_request_row_unwritable",
                 "The durable start-request row could not be written, so the run was "
@@ -541,13 +554,8 @@ def _delegate_start(ctx: ToolContext, prompt: str, max_seconds: Optional[int] =
                                                 invocation_id=invocation_id,
                                                 snapshot_id=("" if recovering else snapshot_id)))
         handle = gateway.start_run(request_body, idempotency_key=invocation_id)
-        # A 202 answers with `jobId` and no `runId` when the run has not bound a run
-        # dir inside the daemon's start timeout; `jobId` is a usable GET/control
-        # handle — discarding it left a live run nobody could wait on or cancel.
         run_id = str(handle.get("runId") or handle.get("jobId") or "")
         if not run_id:
-            # The POST SUCCEEDED, so a run is more likely live here than on the
-            # refusal branch beside it — the registration is retained, not abandoned.
             return _fail("delegate_start", "queued_without_run_id",
                          f"Claudexor returned a queued handle without a run id: {handle!r}",
                          pending_invocation_id=invocation_id,
@@ -601,6 +609,8 @@ def _delegate_start(ctx: ToolContext, prompt: str, max_seconds: Optional[int] =
         config_fingerprint=config_fingerprint, work_order_fingerprint=work_order_fingerprint,
         work_order_coverage=work_order_coverage, work_order_source_request=work_order_source_request,
         authority_fingerprint=authority_fingerprint, snapshot_id=snapshot_id,
+        execution_binding_fingerprint=_start_binding_fingerprint(
+            binding_fingerprint, root, target_root, snapshot_id, resource_ref),
         target_root=target_root, baseline_sha=baseline_sha,
         authority_source=authority_source, resource_ref=resource_ref, processing=processing_info,
         capture_mode=("engine_directory" if resource_ref.get("workspace_kind") == "directory" else

ouroboros/tools/delegate_integration.py

@@ -471,8 +471,9 @@ def _capture_block(entry: _RunCustody, cap_dir: pathlib.Path,
             "the child, a neighbor, or another process may have written there. "
             "The drift is diagnostic evidence; it is not attributed to this child."
             if status == ARTIFACT_STATUS_READY_NO_CHANGES else
-            "NOT APPLIED: the run edited its private execution snapshot only. "
-            "Authority-tree drift was observed while the result was captured; "
+            "NOT APPLIED: the private execution snapshot contains captured changes, "
+            "and authority-tree drift was also observed; "
+            "the author of that drift is unknown. "
             "the existing locked baseline check will decide whether integration "
             "is safe. Nothing reaches the shared tree until explicit disposition."
         )

ouroboros/tools/subagent_integration_delegated.py

@@ -106,6 +106,7 @@ def _capture_at_disposition(
         except (OSError, json.JSONDecodeError, ValueError):
             saved = {}
         if (isinstance(saved, dict) and saved.get("status") == "failed"
+                and saved.get("authority_drift_source_status") == ARTIFACT_STATUS_READY_NO_CHANGES
                 and isinstance(saved.get("authority_drift"), dict)
                 and (saved.get("authority_drift", {}).get("paths")
                      or saved.get("authority_drift", {}).get("error"))):
@@ -521,6 +522,7 @@ def _integrate_git_capture(ctx, entry, decision, reason, manifest, cap_dir, orph
     if decision == "reject":
         drifted_no_change = (
             capture_status == "failed"
+            and manifest.get("authority_drift_source_status") == ARTIFACT_STATUS_READY_NO_CHANGES
             and isinstance(manifest.get("authority_drift"), dict)
             and (manifest["authority_drift"].get("paths")
                  or manifest["authority_drift"].get("error"))

tests/test_delegated_execution_binding.py

@@ -25,6 +25,7 @@ def test_execution_binding_makes_private_root_the_only_write_target():
     assert "/tmp/authority" in text
     assert "sole writable execution root" in text
     assert "read-only identity/reference" in text
+    assert "sha256=" in text
     assert "typed execution-root mismatch" in text
 
 

tests/test_delegated_full_access.py

@@ -125,6 +125,7 @@ def test_full_start_http_contract_and_real_snapshot_capture(full_run):
     assert not (Path(target) / 'native-result.py').exists()
     row = custody.replay(custody.custody_root(ctx))['full-run']
     assert row.access == 'full' and row.project_persistent and row.snapshot_id == key
+    assert row.execution_binding_fingerprint
     row.settled = True
     custody._CUSTODY['full-run'] = row
     capture = delegate._capture_terminal_patch(ctx, row)