diff --git a/ouroboros/delegate_custody.py b/ouroboros/delegate_custody.py index 68067849e..7707ea314 100644 --- a/ouroboros/delegate_custody.py +++ b/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 ""), diff --git a/ouroboros/delegate_registration_policy.py b/ouroboros/delegate_registration_policy.py index dbea0cd26..1564cedad 100644 --- a/ouroboros/delegate_registration_policy.py +++ b/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", diff --git a/ouroboros/delegate_source_coverage.py b/ouroboros/delegate_source_coverage.py index b347ee19e..5dc74d378 100644 --- a/ouroboros/delegate_source_coverage.py +++ b/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, diff --git a/ouroboros/delegate_start_instructions.py b/ouroboros/delegate_start_instructions.py index 2832c870b..b44a26484 100644 --- a/ouroboros/delegate_start_instructions.py +++ b/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( diff --git a/ouroboros/delegate_target_drift.py b/ouroboros/delegate_target_drift.py index ea417d812..33151b7e0 100644 --- a/ouroboros/delegate_target_drift.py +++ b/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 []), diff --git a/ouroboros/tools/delegate.py b/ouroboros/tools/delegate.py index 6468c769c..0196dca5f 100644 --- a/ouroboros/tools/delegate.py +++ b/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 diff --git a/ouroboros/tools/delegate_integration.py b/ouroboros/tools/delegate_integration.py index 7c3f70e97..dfecf90c9 100644 --- a/ouroboros/tools/delegate_integration.py +++ b/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." ) diff --git a/ouroboros/tools/subagent_integration_delegated.py b/ouroboros/tools/subagent_integration_delegated.py index 5290139db..a3c96b240 100644 --- a/ouroboros/tools/subagent_integration_delegated.py +++ b/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")) diff --git a/tests/test_delegated_execution_binding.py b/tests/test_delegated_execution_binding.py index 07ebf996c..171405282 100644 --- a/tests/test_delegated_execution_binding.py +++ b/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 diff --git a/tests/test_delegated_full_access.py b/tests/test_delegated_full_access.py index 6e4eab39a..77c88981d 100644 --- a/tests/test_delegated_full_access.py +++ b/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)