vrr/ouroboros
2
0
Fork 0
mirror of https://github.com/razzant/ouroboros.git synced 2026-10-03 04:07:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

v7next F2: D07 finisher - terminal-evidence leaf, strict worktree registry, sdn facade stands

Browse source 
Owner forks resolved (batch 5, 31.08): 5.9A renamed the deferred terminal
leaf of tools/delegate.py to tools/delegate_terminal_evidence.py (upstream
owns ouroboros/delegate_terminal.py; F5 rename recorded); 5.10A carried the
subagent_worktrees.py strict-registry delta in place with its 11-test pin
suite (tip==merge-base for the module; landed bytes == the reference blob);
5.11A left the subagent_dispatch_notes facade standing.

Ledger rows 3468-3476 cut from tip bytes (drift-probe: 7/9 byte-true,
2 re-emitted; ast=tokens=byte-roundtrip=True on all 9 spans, exit 0) with
the MAXIMAL declared set - every parent-scope call-time read goes through
_delegate() (12 names, LEAVES-pinned). tools/delegate.py 1600 -> 1263
leaves the hard cap and enters the 1001-1500 band; subagent_worktrees.py
1000 -> 1082 enters the band under the owner-sanctioned delta; manifest
regenerated by the official tool. domains.toml untouched (coordinator
seam).

(cherry picked from commit c32e0cd6b304d90e63504e829ff5ee5ebcd1f0fe)
This commit is contained in:
Ouroboros 2026-08-31 09:37:19 +00:00 • committed by Anton Razzhigaev
parent 2878560ed2
commit 1b4a8da957
8 changed files with 907 additions and 381 deletions
Download patch file Download diff file Expand all files Collapse all files

85
docs/v7next/LEDGER_CORRECTIONS.md
View file

@ -1473,3 +1473,88 @@ with evidence, found lane by lane. Applied to the campaign's carried ledger at F
retargeted to the reading leaf module - same oracle retarget-to-owner
shape as the _find_duplicate_task sites; the sentinel's teeth are
restored (the leaf's import-bound name is the one the handler reads).
## From the F2 D07-finisher lane (base 2878560e, 2026-08-31)
1. Scope executed (the three D07 owner forks, decided 31.08 batch 5: 5.9A,
5.10A, 5.11A): the deferred terminal leaf of tools/delegate.py, the
subagent_worktrees.py strict-registry delta with its pin suite, and NO
sdn retirement (the facade stays).
2. F5-RENAME record (owner fork F-2=A, ledger rows 3468-3476): the reference
leaf destination `ouroboros/tools/delegate_terminal.py` is renamed at
landing to `ouroboros/tools/delegate_terminal_evidence.py`. Rationale:
upstream already owns `ouroboros/delegate_terminal.py` ("terminal
reconciliation boundary", 189 lines) and the ledger name would put two
different delegate_terminal modules in neighbouring packages — a
permanent grep/reading trap. Same class as the D01/D03 F5 destination
renames. Rows 3468-3476 read onto the renamed file unchanged otherwise.
3. Terminal leaf landed from tip bytes (rows 3468-3476, D36 handle
`_delegate()`): drift-probe first (reference leaf `--check` against
`git show HEAD:ouroboros/tools/delegate.py`): 7/9 spans byte-true,
_terminal_payload and _delivered_terminal_payload upstream-drifted —
matching the quiet lane's held-back probe evidence — so the leaf was
EMITTED from tip bytes, no oracle semantics replayed. Final proof:
ast=tokens=byte-roundtrip=True on all 9 symbols, leaf_invariants=[],
undeclared_top_level=[], unread_declared=[], exit 0 (re-run after the
manual TYPE_CHECKING preamble addition, the D07-quiet
reconcile-leaf precedent).
4. Declared-set recalc, MAXIMAL form (D10 tools/git precedent, finisher
work-order): the reference cut this leaf with plain preamble imports and
declared only {_emit}; the landed leaf declares EVERY parent-scope name
the moved spans read at call time — 12 names: _Breach,
_PAYLOAD_ENVELOPE_HEADROOM, _emit, _home_isolation_breach,
_preview_payload, _resolve_full_primary_output, _stage_full_output,
_widened_access, add_terminal_source_verification, custody,
home_nested_under_operator_home, tool_result_limit — so patches on the
historical `ouroboros.tools.delegate` surface keep their teeth. Only
stdlib (json) and typing stay preamble imports; annotation-only names
(_Breach for its `-> Optional[_Breach]` use, _RunCustody, ToolContext,
DelegatedRunShape) ride an `if TYPE_CHECKING:` block, inert under future
annotations. New LEAVES row pinned in
tests/test_module_handle_extraction.py.
5. Facade: tools/delegate.py = tip parent - the 9 moved spans (lines
225-576 of the HEAD file) + the grouped EOF re-export block + noqa
discipline: exactly four `# noqa: F401` markers on the import lines of
parent members now read only through `_delegate()` at call time
(_home_isolation_breach, _widened_access, home_nested_under_operator_home,
add_terminal_source_verification — the bindings are load-bearing for the
leaf and must survive ruff F). Every kept def/assign span proven
byte-identical to `git show HEAD:ouroboros/tools/delegate.py` (the diff
of the kept region is exactly those four marker lines); re-exports
proven same-object by import smoke. tools/delegate.py 1600 -> 1263: the
LAST 1600-hard-cap giant of the D07 organ leaves the cap and enters the
1001-1500 band with a rationale. The reference facade-identity rows for this family (held
back by the quiet lane) landed in tests/test_delegate_owner_facades.py
under the renamed leaf.
6. Ф-1 strict-registry delta (rows 1083-1092, owner sanction 5.10A —
SANCTIONED SEMANTIC DELTA in an otherwise byte-preserving lane):
drift-probe first — tip blob of ouroboros/subagent_worktrees.py ==
merge-base 8028f1df blob (fd2db424, upstream never touched the module),
so the reference diff (+104/-22) applied clean; the landed module is
byte-identical to the reference module (blob ee694e4d on both sides).
Semantics: absent registry stays an ordinary empty registry; malformed
registry raises typed SubagentWorktreeRegistryCorrupt for every author/
destructor (provision_worktree, provision_execution_snapshot,
provision_payload_snapshot, find_execution_snapshot,
remove_execution_snapshot, prune_execution_snapshots, remove_worktree,
prune_orphans) instead of silently collapsing to empty; bytes are kept;
one durable subagent_worktree_registry_corrupt event; inspection reads
stay soft; registration moves INSIDE the cleanup scope on all three
provisioning branches. Pin suite
tests/test_subagent_worktree_registry_s6.py copied verbatim from the
oracle (281 lines, 11 tests, red without the delta per D09 entry 10):
imports only stdlib + the module itself, zero v7-only names to reverse-
map; its docstring's sibling reference
(test_delegated_skill_payload.py::test_registry_save_failure_leaves_no_orphan_snapshot_dir)
exists on tip; the oracle registered it in no conftest path-keyed table.
The one pre-existing tip test touching the registry
(tests/test_acting_subagents.py:1298) uses the soft read, whose
signature and behavior are unchanged.
7. Ф-3 (sdn): no action, per owner 5.11A — the quiet lane's entry 8 stands
(rows 3937-3938 satisfied as identity; retirement stays an F5
consumer-rebind item).
8. Ratchet (official regenerator): ouroboros/tools/delegate.py enters the
band by extraction (1600->1263, rationale recorded);
ouroboros/subagent_worktrees.py enters the band by the sanctioned delta
(1000->1082, rationale recorded). domains.toml untouched (coordinator
seam owns the map).

2
ouroboros/size_ratchet_manifest.py
View file

@ -144,12 +144,14 @@ BAND_PATHS = {
"ouroboros/safety.py": "Entered the band from 954 lines with the safety-supervisor rate-limit fix: ONE shared model-call helper now serves both the primary and repair safety calls (it already deletes the duplicated call block), recognising a provider rate limit in BOTH wire shapes, taking one bounded deadline-capped backoff plus one retry, then blocking that one call with the typed non-verdict SAFETY_UNAVAILABLE outcome plus a durable audit row (a short storm latch answers further checks in the window without provider calls); the bounded newest-first conversation budget is the second half.",
"ouroboros/skill_review_runner.py": None,
"ouroboros/subagent_runtime.py": "Configured-retry refusals mirrored typed (triad 2026-08-30) push the module just over 1000; no new subsystem, same seam.",
"ouroboros/subagent_worktrees.py": "Owner-sanctioned strict-registry delta (v7 rows 1083-1092, fork F-1=A) grew the module 1000->1082: typed refusal of a malformed registry instead of silent collapse-to-empty; shrink-only direction",
"ouroboros/subagents.py": "D07 split brought the dispatch monolith DOWN from 1593 into the band (->1370); route-health family extracted to subagent_route_health.py, shrink-only direction",
"ouroboros/task_results.py": "Authority reads need an explicit strict mode so malformed child records cannot become a false zero count.",
"ouroboros/task_status.py": None,
"ouroboros/tools/browser.py": None,
"ouroboros/tools/commit_gate.py": "Grew INTO the band by the review-wave fix binding the actor reference (delivery class) into the commit review contract fingerprint \u2014 same-module contract identity, splitting it would separate the fingerprint from its gate.",
"ouroboros/tools/core.py": "D05 ledger split (rows 311-349): read/list and owner-chat delivery spans moved to core_file_tools/core_artifacts; facade re-enters the band from above (2283 -> 1373) and shrinks further when the residual catalog split lands",
"ouroboros/tools/delegate.py": "D07 finisher DEL1 split brought the nanny-verb monolith DOWN from the 1600 hard cap into the band (1600->1263); terminal-evidence family extracted to tools/delegate_terminal_evidence.py, shrink-only direction",
"ouroboros/tools/plan_review_runtime.py": "Entered the band from 986 lines: timeout custody synthesis joined the existing plan-review runtime owner while preserving profile-continuity disclosures and typed health facts during target integration.",
"ouroboros/tools/review_context_atlas.py": "Grew INTO the band by the #284 pack-arithmetic fixes: measured render charged at admission, exact per-row costs, target capped at the hard rail, honest eviction diagnostics \u2014 all in the module that owns the arithmetic.",
"ouroboros/tools/skill_exec.py": None,

126
ouroboros/subagent_worktrees.py
View file

@ -16,6 +16,7 @@ from __future__ import annotations
import contextlib
import json
import logging
import os
import re
import shutil
@ -32,6 +33,8 @@ from ouroboros.utils import atomic_write_json
from ouroboros.config import DATA_DIR, get_subagent_projects_root, get_subagent_worktree_root
from ouroboros.retention import age_cutoff, get_gc_retention_days
log = logging.getLogger(__name__)
_REGISTRY_NAME = "subagent_worktrees.json"
_LOCK_NAME = ".worktree_ops.lock"
_LOCK_TIMEOUT_SEC = 120.0
@ -99,16 +102,65 @@ def _safe_name(task_id: Any) -> str:
return safe
def _load_registry(data_dir: Optional[Any] = None) -> List[Dict[str, Any]]:
class SubagentWorktreeRegistryCorrupt(RuntimeError):
"""``state/subagent_worktrees.json`` exists but cannot be read as a registry.
Raised by every caller that would go on to REWRITE the file. Collapsing a
malformed registry to an empty one hands the next write a clean slate: the
rows are gone, and with them the only record naming the checkouts and the
``refs/ouroboros/delegated/*`` refs those rows pin — a leak nothing can
reconcile afterwards. The malformed bytes are kept instead, which is what
the sibling registries (cancel intents, terminal deliveries) do and what
the startup GC already does when the custody log is unreadable.
"""
def _load_registry(
data_dir: Optional[Any] = None, *, strict: bool = False, op: str = "",
) -> List[Dict[str, Any]]:
"""The registered rows; ``strict`` refuses a malformed registry.
ABSENT is an ordinary empty registry in both modes (the first-write case).
MALFORMED is a different fact, and ``strict=True`` — passed by everything
that authors a record or acts destructively on one — reports it as such
instead of as "nothing is registered". Inspection reads (the UI listing)
stay soft: they display what they can and destroy nothing.
"""
path = _registry_path(data_dir)
if not path.is_file():
return []
try:
raw = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError, ValueError):
entries = raw.get("worktrees") if isinstance(raw, dict) else raw
if not isinstance(entries, list):
raise ValueError("subagent worktree registry 'worktrees' is not a list")
except (OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
if strict:
raise _refuse_corrupt_registry(data_dir, op, exc) from exc
return []
entries = raw.get("worktrees") if isinstance(raw, dict) else raw
if isinstance(entries, list):
return [e for e in entries if isinstance(e, dict)]
return []
return [e for e in entries if isinstance(e, dict)]
def _refuse_corrupt_registry(
data_dir: Optional[Any], op: str, exc: Exception,
) -> SubagentWorktreeRegistryCorrupt:
"""Disclose an unreadable registry durably, then refuse the mutation."""
path = _registry_path(data_dir)
log.error(
"subagent worktree registry is corrupt; %s refused, bytes kept (%s)",
op or "mutation", exc,
)
try:
from ouroboros.utils import append_jsonl, utc_now_iso
append_jsonl(
_data_dir(data_dir) / "logs" / "events.jsonl",
{"ts": utc_now_iso(), "type": "subagent_worktree_registry_corrupt",
"op": str(op or ""), "registry": str(path), "error": str(exc)[:200]},
)
except Exception:
log.debug("registry-corrupt event append failed", exc_info=True)
return SubagentWorktreeRegistryCorrupt(str(exc))
def _save_registry(entries: List[Dict[str, Any]], data_dir: Optional[Any] = None) -> None:
@ -259,9 +311,21 @@ def provision_worktree(
created_at=time.time(),
parent_task_id=str(parent_task_id or ""),
)
entries = [e for e in _load_registry(data_dir) if e.get("path") != str(wt_path)]
entries.append(asdict(handle))
_save_registry(entries, data_dir)
try:
entries = [
e for e in _load_registry(data_dir, strict=True, op="provision_worktree")
if e.get("path") != str(wt_path)
]
entries.append(asdict(handle))
_save_registry(entries, data_dir)
except Exception:
# Registration is INSIDE the cleanup scope, mirroring the snapshot
# branches below: a worktree nothing registered is invisible to
# disposal and retention, so a corrupt registry (strict load) or a
# failed write would otherwise strand the checkout AND its branch
# on every retry, without bound.
_remove_paths(repo_dir, wt_path, branch, allowed_root=root)
raise
return handle
@ -529,12 +593,24 @@ def provision_execution_snapshot(
entry_count=entry_count,
excluded_untracked=tuple(excluded),
)
entries = [e for e in _load_registry(data_dir) if e.get("path") != str(wt_path)]
record = asdict(handle)
record["kind"] = _KIND_DELEGATED_EXEC
record["excluded_untracked"] = excluded
entries.append(record)
_save_registry(entries, data_dir)
try:
entries = [
e for e in _load_registry(data_dir, strict=True, op="provision_execution_snapshot")
if e.get("path") != str(wt_path)
]
record = asdict(handle)
record["kind"] = _KIND_DELEGATED_EXEC
record["excluded_untracked"] = excluded
entries.append(record)
_save_registry(entries, data_dir)
except Exception:
# Registration is INSIDE the cleanup scope, exactly like the payload
# branch: a snapshot nothing registered is invisible to disposal and
# retention, and on this branch it also strands the baseline ref,
# which pins its commit against git's own GC for good.
_remove_paths(target, wt_path, "", allowed_root=root)
_git(target, "update-ref", "-d", baseline_ref, check=False)
raise
return handle
@ -820,7 +896,10 @@ def provision_payload_snapshot(
)
# Registry write INSIDE the cleanup scope: an unregistered snapshot
# directory would be invisible to disposal/retention (orphan leak).
entries = [e for e in _load_registry(data_dir) if e.get("path") != str(wt_path)]
entries = [
e for e in _load_registry(data_dir, strict=True, op="provision_payload_snapshot")
if e.get("path") != str(wt_path)
]
record = asdict(handle)
record["kind"] = _KIND_DELEGATED_EXEC
record["excluded_untracked"] = []
@ -837,7 +916,7 @@ def find_execution_snapshot(snapshot_id: str, data_dir: Optional[Any] = None) ->
snap = str(snapshot_id or "").strip()
if not snap:
return None
for entry in _load_registry(data_dir):
for entry in _load_registry(data_dir, strict=True, op="find_execution_snapshot"):
if entry.get("kind") == _KIND_DELEGATED_EXEC and entry.get("snapshot_id") == snap:
return entry
return None
@ -877,7 +956,7 @@ def remove_execution_snapshot(
_git(target, "update-ref", "-d", ref, check=False)
except Exception:
pass
survivors = [e for e in _load_registry(data_dir) if not (
survivors = [e for e in _load_registry(data_dir, strict=True, op="remove_execution_snapshot") if not (
e.get("kind") == _KIND_DELEGATED_EXEC and e.get("snapshot_id") == entry.get("snapshot_id")
)]
_save_registry(survivors, data_dir)
@ -902,7 +981,7 @@ def prune_execution_snapshots(
open_ids = {str(s) for s in (open_snapshot_ids or set())}
removed: List[str] = []
kept: List[str] = []
for entry in list(_load_registry(data_dir)):
for entry in list(_load_registry(data_dir, strict=True, op="prune_execution_snapshots")):
if entry.get("kind") != _KIND_DELEGATED_EXEC:
continue
snap = str(entry.get("snapshot_id") or "")
@ -924,7 +1003,7 @@ def remove_worktree(
) -> bool:
"""Tear down a worktree by task_id or path; unregister it. Returns success."""
want_path = str(Path(path).resolve()) if path else ""
entries = _load_registry(data_dir)
entries = _load_registry(data_dir, strict=True, op="remove_worktree")
match: Optional[Dict[str, Any]] = None
for entry in entries:
if task_id and entry.get("task_id") == str(task_id):
@ -937,7 +1016,10 @@ def remove_worktree(
with _ops_lock(root):
if match is not None:
_remove_paths(Path(match.get("repo_dir") or "."), Path(match.get("path") or ""), match.get("branch") or "", allowed_root=root)
survivors = [e for e in _load_registry(data_dir) if e.get("path") != match.get("path")]
survivors = [
e for e in _load_registry(data_dir, strict=True, op="remove_worktree")
if e.get("path") != match.get("path")
]
_save_registry(survivors, data_dir)
return True
# Unregistered path: best-effort directory removal, but ONLY inside the
@ -965,7 +1047,7 @@ def prune_orphans(
kept: List[Dict[str, Any]] = []
repos: set[str] = set()
with _ops_lock(root):
for entry in _load_registry(data_dir):
for entry in _load_registry(data_dir, strict=True, op="prune_orphans"):
if entry.get("kind") == _KIND_DELEGATED_EXEC:
# Delegated execution snapshots have their OWN lifecycle: they persist
# until the run's explicit patch disposition, and the startup GC

375
ouroboros/tools/delegate.py
View file

@ -43,7 +43,7 @@ from ouroboros.subagent_work_order import ( # noqa: F401 - compatibility re-exp
assignment_instructions as _assignment_instructions,
)
from ouroboros.delegate_source_coverage import (
add_terminal_source_verification,
add_terminal_source_verification, # noqa: F401 (leaf reads it through _delegate() at call time)
prepare_work_order_start_binding,
record_started_custody,
)
@ -136,9 +136,9 @@ _TERMINAL_STATES = custody.TERMINAL_STATES
from ouroboros.delegate_containment import ( # noqa: E402
_ACCESS_UNVERIFIED, # noqa: F401 (re-export: tests address it through this module)
_Breach,
_home_isolation_breach,
_widened_access,
home_nested_under_operator_home,
_home_isolation_breach, # noqa: F401 (leaf reads it through _delegate() at call time)
_widened_access, # noqa: F401 (leaf reads it through _delegate() at call time)
home_nested_under_operator_home, # noqa: F401 (leaf reads it through _delegate() at call time)
)
_POLL_INTERVAL_SEC = 3.0
# Claudexor's own schema bound on maxSeconds (packages/schema/src/control.ts).
@ -222,358 +222,6 @@ def _presence_delegate_read_refusal(ctx: ToolContext) -> str:
)
def _containment_breach(detail: Dict[str, Any], authority: "DelegatedRunShape") -> Optional[_Breach]:
"""Everything the ENGINE enforced, checked against what the host asked for.
ONE reader for both halves of containment — the access profile and the harness
HOME — because they fail identically: the request is only a request, the engine
derives the truth, and a verification written for one half leaves the other
trusting an echo. The HOME half is asked only of a run that carried the marker;
a read-only child is scoped by Claudexor's ordinary envelope and asks for nothing.
"""
widened = _widened_access(detail, authority.access)
if widened:
return _Breach(
"access_profile_widened",
f"The delegated run was enforced at access profile {widened!r} while this "
f"task is only entitled to {authority.access!r}.",
{"entitled_access": authority.access, "effective_access": widened},
)
if authority.delegated:
return _home_isolation_breach(detail)
return None
_NESTED_HOME_NOTE = (
"The scoped harness HOME for this run sits INSIDE the operator's own home, which is "
"where the engine roots its scoped homes. That is allowed and the run's work is usable, "
"but it is not isolation from the operator's home: everything there — credential stores "
"and the Claudexor daemon token included — stays readable at its absolute path. Do NOT "
"describe this run as running in an isolated home"
)
_NO_BOUNDARY_NOTE = (
"NO OS-ENFORCED BOUNDARY was applied to this run. The engine reported no confinement "
"mechanism for it, so the only containment it had is a scoped HOME — a redirect of "
"`~`-relative lookups, which leaves the operator's home, credential stores and the "
"Claudexor daemon token readable at their absolute paths. The run was allowed and its "
"work is usable; do NOT describe it as sandboxed, confined or isolated, and weigh its "
"output as coming from an unconfined shell in this worktree"
)
def _containment_evidence(detail: Dict[str, Any]) -> Dict[str, Any]:
"""What the ARTIFACTS prove about this run's containment — never what was asked.
DESTINATION 3 of the disclosure: this is what the nanny hands its parent.
BOTH halves, in one reader, because a report that states only the scoped HOME is the
defect this function was rewritten to remove: a run with a kernel-enforced boundary
and a run with none produced BYTE-IDENTICAL evidence here, both reading
``verified: true`` with a note about the HOME. Claudexor's own confinement document
says the scoped home "is not a boundary and must never be reported as one".
The predicate is what the engine says it APPLIED (``confinement_mechanism`` plus the
denied path it proved), never which OS this host is. Ouroboros does not know what the
engine did — only the artifact does — and a platform test would additionally freeze
today's answer: the day a boundary ships for another OS, this reader is already right.
Judged by the SAME predicate that halts a breached run, not by having been reached
after it: a report whose honesty depends on its call site is one refactor away from
claiming a containment nobody checked.
This is also where a MISSING fact lands, because it is a reporting question and not an
enforcement one: an attempt that disclosed nothing proves nothing, so ``verified``
stays false and ``disclosed`` says how much of the run is actually covered. Silence
read as success and silence enforced as a fault are the two ways to be wrong here,
and stating the count avoids both.
"""
from ouroboros.gateways.claudexor import attempt_containment
attempts = attempt_containment(str(custody.summary_of(detail).get("runDir") or ""))
disclosed = sum(1 for attempt in attempts if attempt.home_isolated is not None)
# An engine that reported nothing is indistinguishable from one that applied nothing,
# and the mechanisms the ATTEMPTS name are the vocabulary — Ouroboros keeps no list of
# its own to fall out of date. "Every attempt" and not "any": one unconfined attempt
# is an unconfined run.
mechanisms = sorted({attempt.boundary_mechanism for attempt in attempts})
boundary = mechanisms[0] if attempts and len(mechanisms) == 1 and mechanisms[0] else ""
# A3: the engine's own typed reason for a missing boundary — an AMPLIFIER of
# the unconfined disclosure (why there is no mechanism on this host), parsed
# from the same attempt artifact. Telemetry only, never an admission token.
unavailable_reasons = sorted({
attempt.confinement_unavailable_reason
for attempt in attempts if attempt.confinement_unavailable_reason
})
# A3: a scoped home NESTED under the operator's own is allowed (the engine's
# own layout — disclosed, never refused), but it is NOT "outside the
# operator's own": the daemon token stays reachable at its absolute path.
# Recorded on the report and honoured by every branch below, so a run that
# ALSO carries an OS boundary can no longer be promoted to verified with a
# note that contradicts its own artifact — and so `_record_containment` keeps
# emitting the durable unconfined row for it.
nested = home_nested_under_operator_home(detail)
report = {"verified": False, "attempts": len(attempts), "disclosed": disclosed,
"os_boundary": boundary, "nested_under_operator_home": nested}
if unavailable_reasons:
report["confinement_unavailable_reason"] = "; ".join(unavailable_reasons)
breach = _home_isolation_breach(detail)
if breach is not None:
return {**report, "note": breach.detail}
if not disclosed:
return {**report, "note":
"this run recorded no harness-HOME fact, so its confinement is UNPROVEN "
"— do not report it as isolated"}
if disclosed < len(attempts):
return {**report, "note":
"not every attempt of this run recorded a harness-HOME fact, so its "
"confinement is UNPROVEN — do not report it as isolated"}
if nested:
note = _NESTED_HOME_NOTE
if boundary:
note += (
f" (an {boundary} boundary WAS applied — weigh it as the real containment, "
"but the scoped HOME is not one)"
)
if unavailable_reasons:
note += " (engine-declared reason: " + "; ".join(unavailable_reasons) + ")"
return {**report, "note": note}
if not boundary:
note = _NO_BOUNDARY_NOTE
if unavailable_reasons:
note += (
" (engine-declared reason: " + "; ".join(unavailable_reasons) + ")"
)
return {**report, "note": note}
return {**report, "verified": True, "note":
f"every attempt recorded a scoped harness HOME outside the operator's own AND "
f"an applied {boundary} boundary, proven against a path it denies"}
def _terminal_payload(run_id: str, detail: Dict[str, Any],
authority: "DelegatedRunShape") -> Dict[str, Any]:
summary = custody.summary_of(detail)
payload = {
"status": "terminal",
"run_id": run_id,
"state": str(summary.get("state") or ""),
# The APPLIED model, from the engine's own summary — '' when the run
# never disclosed one (live unpinned runs really do), shown as absence
# rather than the requested model dressed up as the applied one.
"model": str(summary.get("model") or ""),
"outcome_banner": detail.get("outcomeBanner"),
"outcome_facts": summary.get("outcomeFacts"),
"output_conformance": summary.get("outputConformance"),
"final_summary": detail.get("finalSummary"),
"primary_output": detail.get("primaryOutput"),
"failure": summary.get("failure"),
"last_seq": int(detail.get("lastSeq") or 0),
"cost": _reported_cost(summary),
# The ACCESS half of the same honesty, on EVERY terminal payload — see
# `_access_evidence`. Both lanes: `readonly` staying `readonly` is the profile
# that matters most, while `containment` is asked only of marker-carrying runs.
"access_evidence": _access_evidence(detail, authority.access),
}
if authority.delegated:
payload["containment"] = _containment_evidence(detail)
facts = payload.get("outcome_facts")
if isinstance(facts, dict) and str(facts.get("reason") or "") == "input_required":
# The codex-shaped question (B4): that lane has no mid-run channel, so a
# question arrives as this TERMINAL. There is deliberately NO rerun verb
# here — the engine's rerun_with_feedback would start a run outside this
# task's custody trail — so the honest answer path is a plain new start.
payload["input_required_note"] = (
"This run ended NEEDING INPUT (outcome_facts.reason=input_required — "
"see outcome_facts.work_state.required_inputs). Its harness has no "
"mid-run question channel, so the question arrives as this terminal. Answer it by "
"starting a plain NEW delegate_start(subagent_id=..., prompt=...) whose "
"prompt carries the original "
"assignment plus the answers; custody of the new run stays with you. "
"Do not look for a rerun/decision verb — none exists on this surface."
)
return payload
def _access_evidence(detail: Dict[str, Any], expected: str) -> Dict[str, Any]:
"""What the engine's own DERIVED profile proves about this finished run.
``effectiveAccess`` is the only witness: ``summary["access"]`` is computed as
``effectiveAccess ?? the client's own request``, so reading it compares the request
against itself and always passes. A WIDER profile is already a breach before this
runs; an ABSENT one cannot be enforced on a run that is over — cancelling a
succeeded run to punish missing evidence would destroy the result the lane exists
to fetch (the v6.87.37 lesson) — so it is named here instead.
"""
summary = custody.summary_of(detail)
effective = str(summary.get("effectiveAccess") or "")
state = str(summary.get("state") or "")
report = {"requested": expected, "effective": effective,
"verified": bool(effective), "state": state}
if effective:
return report
if state in custody.SUCCEEDED_STATES:
return {**report, "note":
"this run SUCCEEDED without ever disclosing an effective access "
f"profile, so there is no evidence the engine enforced {expected!r} — "
"do not report its containment as verified"}
return {**report, "note":
"no effective access profile was disclosed; a run that did not succeed may "
"never have had one, so this is absence of evidence, not a breach"}
def _record_containment(ctx: ToolContext, entry: Optional[_RunCustody],
payload: Dict[str, Any]) -> None:
"""DESTINATION 1 of the disclosure: the durable record, written once per run.
A missing boundary is not a fault and produces no refusal, which is exactly why it
needs a durable line of its own — the run succeeds, its patch is integrated, and
nothing else in the record would ever say the work came out of an unconfined shell.
Emitted from what the PARENT was told, so the two cannot disagree.
"Once per run" is now a DURABLE fact rather than a process-local one: the custody
entry is replayed from the event log, so a restarted worker polling an already
terminal run does not append a second identical finding.
A NESTED scoped home is disclosed even when an OS boundary WAS recorded (A3):
the boundary is real containment, the scoped home is not, and suppressing the
row for that shape left the one durable line that says "this ran with the
operator's home reachable" unwritten.
"""
containment = payload.get("containment")
if not isinstance(containment, dict):
return
if containment.get("os_boundary") and not containment.get("nested_under_operator_home"):
return
if entry is not None and entry.containment_disclosed:
return
_emit(ctx, custody.UNCONFINED, {
"run_id": entry.run_id if entry is not None else "",
"route": entry.route_id if entry is not None else "",
"state": str(payload.get("state") or ""),
"os_boundary": str(containment.get("os_boundary") or ""),
"attempts": containment.get("attempts"),
"home_disclosed": containment.get("disclosed"),
"nested_under_operator_home": bool(containment.get("nested_under_operator_home")),
"note": containment.get("note"),
**({"confinement_unavailable_reason": containment["confinement_unavailable_reason"]}
if containment.get("confinement_unavailable_reason") else {}),
})
if entry is not None:
entry.containment_disclosed = True
def _reported_cost(summary: Dict[str, Any]) -> Dict[str, Any]:
"""What this run cost, as the AGENT will read it.
This is the payload the nanny relays to its parent, so it must tell the same story
the ledger does. It used to hardcode `$0.00 / final` — the exact shape the settlement
fix was written to eliminate — so a run that really charged money settled honestly in
the ledger and then told the reasoning path the work was free.
"""
spend, estimated = custody.disclosed_spend(summary)
if spend is None:
return {
"cost_usd": None,
"cost_final": False,
"note": "the harness disclosed no spend for this run; treat the cost as UNKNOWN, not zero",
}
if estimated:
# The amount is the best fact anyone has, so it rides; the FINALITY does not. An
# estimated zero is not a proven free session and an estimated charge is not a
# closed book — both are `cost_final: False`, matching the ledger row exactly.
return {
"cost_usd": spend,
"cost_final": False,
"note": "the harness ESTIMATED this run's spend rather than settling it; treat "
"the amount as APPROXIMATE and the cost as NOT final",
}
if spend > 0:
return {
"cost_usd": spend,
"cost_final": True,
"note": "this run was BILLED — it did not ride the subscription",
}
return {
"cost_usd": 0.0,
"cost_final": True,
"note": "subscription session — already paid; the nanny's own model calls are metered separately",
}
# -- output delivery -----------------------------------------------------------
def _delivered_terminal_payload(ctx: ToolContext, run_id: str, detail: Dict[str, Any],
authority: "DelegatedRunShape",
entry: Optional[_RunCustody] = None,
gateway: Any = None) -> Dict[str, Any]:
"""The terminal payload, delivered whole or declared partial — never head-cut.
``final_summary``/``primary_output`` carry the run's real work product, and Claudexor
returns a preview of up to 256 KiB. Outer truncation would head-cut that at the tool
result limit and sever the JSON mid-string, which destroys the document rather than
shortening it. So the payload bounds ITSELF against the same limit the truncator
applies, and the remainder becomes a readable artifact — after the engine's bounded
preview has been resolved to the verified full artifact, because a payload built on
a truncated preview delivers 256 KiB wearing the whole result's name.
"""
full = _terminal_payload(run_id, detail, authority)
if entry is not None:
add_terminal_source_verification(full, entry)
# Requested-vs-applied model, the review lane's own lexicon and rule
# (AgentSessionReviewExecutor): compared only when BOTH are non-empty —
# the engine writes aliases ('sonnet' beside 'claude-opus-5'), so a
# mismatch is an advisory disclosure, never a failure of the run.
requested_model = str(getattr(entry, "model", "") or "") if entry is not None else ""
applied_model = str(full.get("model") or "")
if requested_model and applied_model and requested_model != applied_model:
full["capability_delta"] = [{
"kind": "capability_delta",
"requested": f"model {requested_model}",
"effective": f"model {applied_model}",
"reason": "session_route_resolves_its_own_model",
}]
primary, full_ok, full_note = _resolve_full_primary_output(
gateway, run_id, full.get("primary_output"))
full["primary_output"] = primary
budget = tool_result_limit("delegate_wait")
text = json.dumps(full, ensure_ascii=False, indent=2)
if len(text) <= budget - _PAYLOAD_ENVELOPE_HEADROOM:
full["output_delivery"] = {
# An unresolved engine-side truncation makes even an inline-fitting payload
# NOT the whole result: complete/consumed follow the verified fact.
"complete": full_ok, "consumed": full_ok, "inline_is_preview": False,
"total_chars": len(text), "artifact": None, "read_next": None,
"note": ("The whole terminal payload is inline." if full_ok else
"INLINE BUT INCOMPLETE AT THE SOURCE: the engine reported its "
"primary output as a bounded preview and the full artifact could "
"not be matched to the size or the preview the run itself reported "
"(see primary_output_full). Treat this "
"as incomplete evidence, not as the verdict."),
}
if full_note is not None:
full["output_delivery"]["primary_output_full"] = full_note
return full
artifact = _stage_full_output(ctx, run_id, text)
_emit(ctx, custody.OUTPUT_SPILLED, {"run_id": run_id, "total_chars": len(text),
"artifact": (artifact or {}).get("path", ""),
"bytes": (artifact or {}).get("bytes"),
"sha256": (artifact or {}).get("sha256", ""),
"staged": artifact is not None,
"full_content": bool(full_ok and artifact is not None)})
if entry is not None and artifact is not None:
if entry.output_consumed and entry.output_sha and artifact["sha256"] != entry.output_sha:
# The ack named OTHER bytes: a re-stage of different content at the same
# path owes a fresh acknowledgement — consumed never transfers by path.
entry.output_consumed = False
entry.output_sha = artifact["sha256"]
entry.output_artifact = artifact["path"]
entry.output_complete = bool(full_ok)
return _preview_payload(full, text, artifact, budget,
consumed=bool(entry is not None and entry.output_consumed),
full_ok=full_ok, full_note=full_note)
# -- tools --------------------------------------------------------------------
@ -1598,3 +1246,18 @@ def get_tools() -> List[ToolEntry]:
__all__ = ["get_tools"]
# v7next F2 (D07): moved spans live in their owner leaf; re-exported here
# so this facade stays the single import surface for callers and tests.
from ouroboros.tools.delegate_terminal_evidence import ( # noqa: E402, F401 -- intentional public re-exports
_NESTED_HOME_NOTE,
_NO_BOUNDARY_NOTE,
_access_evidence,
_containment_breach,
_containment_evidence,
_delivered_terminal_payload,
_record_containment,
_reported_cost,
_terminal_payload,
)

392
ouroboros/tools/delegate_terminal_evidence.py Normal file
View file

@ -0,0 +1,392 @@
"""The terminal story of ONE delegated run, as the parent reads it.
Containment breach detection and evidence, the terminal payload with its access
evidence and reported cost, and whole-or-declared-partial delivery. Extracted
from ``ouroboros/tools/delegate.py`` at its size gate (v7 DEL1 split);
``tools.delegate`` re-exports every name (same objects), so the wait loop, the
tests and monkeypatch targets keep addressing them on THAT surface.
The v7 ledger named this leaf ``tools/delegate_terminal.py``; it lands as
``delegate_terminal_evidence.py`` because upstream already owns
``ouroboros/delegate_terminal.py`` (the terminal reconciliation boundary) and
two neighbouring modules answering to one name would be a permanent grep trap
(owner fork F-2=A; the rename is recorded in the carried ledger).
Every parent-scope name the moved bodies read at call time is DECLARED and read
through ``_delegate()`` — the parent monolith keeps the one rebindable binding
per member, so patches on the historical surface keep their teeth.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any, Dict, Optional
if TYPE_CHECKING: # pragma: no cover - annotation-only names, lazy under future annotations
from ouroboros.delegate_containment import _Breach
from ouroboros.delegate_custody import RunCustody as _RunCustody
from ouroboros.subagents import DelegatedRunShape
from ouroboros.tools.registry import ToolContext
def _delegate():
"""The parent module, read at call time.
The parent owns the rebindable module state and the members tests
monkeypatch there; reading them through the module at each call keeps
one binding, where a from-import would freeze the value this leaf saw
at import time (the owner-approved D18/D33 mechanical exception).
"""
from ouroboros.tools import delegate
return delegate
def _containment_breach(detail: Dict[str, Any], authority: "DelegatedRunShape") -> Optional[_Breach]:
"""Everything the ENGINE enforced, checked against what the host asked for.
ONE reader for both halves of containment — the access profile and the harness
HOME — because they fail identically: the request is only a request, the engine
derives the truth, and a verification written for one half leaves the other
trusting an echo. The HOME half is asked only of a run that carried the marker;
a read-only child is scoped by Claudexor's ordinary envelope and asks for nothing.
"""
widened = _delegate()._widened_access(detail, authority.access)
if widened:
return _delegate()._Breach(
"access_profile_widened",
f"The delegated run was enforced at access profile {widened!r} while this "
f"task is only entitled to {authority.access!r}.",
{"entitled_access": authority.access, "effective_access": widened},
)
if authority.delegated:
return _delegate()._home_isolation_breach(detail)
return None
_NESTED_HOME_NOTE = (
"The scoped harness HOME for this run sits INSIDE the operator's own home, which is "
"where the engine roots its scoped homes. That is allowed and the run's work is usable, "
"but it is not isolation from the operator's home: everything there — credential stores "
"and the Claudexor daemon token included — stays readable at its absolute path. Do NOT "
"describe this run as running in an isolated home"
)
_NO_BOUNDARY_NOTE = (
"NO OS-ENFORCED BOUNDARY was applied to this run. The engine reported no confinement "
"mechanism for it, so the only containment it had is a scoped HOME — a redirect of "
"`~`-relative lookups, which leaves the operator's home, credential stores and the "
"Claudexor daemon token readable at their absolute paths. The run was allowed and its "
"work is usable; do NOT describe it as sandboxed, confined or isolated, and weigh its "
"output as coming from an unconfined shell in this worktree"
)
def _containment_evidence(detail: Dict[str, Any]) -> Dict[str, Any]:
"""What the ARTIFACTS prove about this run's containment — never what was asked.
DESTINATION 3 of the disclosure: this is what the nanny hands its parent.
BOTH halves, in one reader, because a report that states only the scoped HOME is the
defect this function was rewritten to remove: a run with a kernel-enforced boundary
and a run with none produced BYTE-IDENTICAL evidence here, both reading
``verified: true`` with a note about the HOME. Claudexor's own confinement document
says the scoped home "is not a boundary and must never be reported as one".
The predicate is what the engine says it APPLIED (``confinement_mechanism`` plus the
denied path it proved), never which OS this host is. Ouroboros does not know what the
engine did — only the artifact does — and a platform test would additionally freeze
today's answer: the day a boundary ships for another OS, this reader is already right.
Judged by the SAME predicate that halts a breached run, not by having been reached
after it: a report whose honesty depends on its call site is one refactor away from
claiming a containment nobody checked.
This is also where a MISSING fact lands, because it is a reporting question and not an
enforcement one: an attempt that disclosed nothing proves nothing, so ``verified``
stays false and ``disclosed`` says how much of the run is actually covered. Silence
read as success and silence enforced as a fault are the two ways to be wrong here,
and stating the count avoids both.
"""
from ouroboros.gateways.claudexor import attempt_containment
attempts = attempt_containment(str(_delegate().custody.summary_of(detail).get("runDir") or ""))
disclosed = sum(1 for attempt in attempts if attempt.home_isolated is not None)
# An engine that reported nothing is indistinguishable from one that applied nothing,
# and the mechanisms the ATTEMPTS name are the vocabulary — Ouroboros keeps no list of
# its own to fall out of date. "Every attempt" and not "any": one unconfined attempt
# is an unconfined run.
mechanisms = sorted({attempt.boundary_mechanism for attempt in attempts})
boundary = mechanisms[0] if attempts and len(mechanisms) == 1 and mechanisms[0] else ""
# A3: the engine's own typed reason for a missing boundary — an AMPLIFIER of
# the unconfined disclosure (why there is no mechanism on this host), parsed
# from the same attempt artifact. Telemetry only, never an admission token.
unavailable_reasons = sorted({
attempt.confinement_unavailable_reason
for attempt in attempts if attempt.confinement_unavailable_reason
})
# A3: a scoped home NESTED under the operator's own is allowed (the engine's
# own layout — disclosed, never refused), but it is NOT "outside the
# operator's own": the daemon token stays reachable at its absolute path.
# Recorded on the report and honoured by every branch below, so a run that
# ALSO carries an OS boundary can no longer be promoted to verified with a
# note that contradicts its own artifact — and so `_record_containment` keeps
# emitting the durable unconfined row for it.
nested = _delegate().home_nested_under_operator_home(detail)
report = {"verified": False, "attempts": len(attempts), "disclosed": disclosed,
"os_boundary": boundary, "nested_under_operator_home": nested}
if unavailable_reasons:
report["confinement_unavailable_reason"] = "; ".join(unavailable_reasons)
breach = _delegate()._home_isolation_breach(detail)
if breach is not None:
return {**report, "note": breach.detail}
if not disclosed:
return {**report, "note":
"this run recorded no harness-HOME fact, so its confinement is UNPROVEN "
"— do not report it as isolated"}
if disclosed < len(attempts):
return {**report, "note":
"not every attempt of this run recorded a harness-HOME fact, so its "
"confinement is UNPROVEN — do not report it as isolated"}
if nested:
note = _NESTED_HOME_NOTE
if boundary:
note += (
f" (an {boundary} boundary WAS applied — weigh it as the real containment, "
"but the scoped HOME is not one)"
)
if unavailable_reasons:
note += " (engine-declared reason: " + "; ".join(unavailable_reasons) + ")"
return {**report, "note": note}
if not boundary:
note = _NO_BOUNDARY_NOTE
if unavailable_reasons:
note += (
" (engine-declared reason: " + "; ".join(unavailable_reasons) + ")"
)
return {**report, "note": note}
return {**report, "verified": True, "note":
f"every attempt recorded a scoped harness HOME outside the operator's own AND "
f"an applied {boundary} boundary, proven against a path it denies"}
def _terminal_payload(run_id: str, detail: Dict[str, Any],
authority: "DelegatedRunShape") -> Dict[str, Any]:
summary = _delegate().custody.summary_of(detail)
payload = {
"status": "terminal",
"run_id": run_id,
"state": str(summary.get("state") or ""),
# The APPLIED model, from the engine's own summary — '' when the run
# never disclosed one (live unpinned runs really do), shown as absence
# rather than the requested model dressed up as the applied one.
"model": str(summary.get("model") or ""),
"outcome_banner": detail.get("outcomeBanner"),
"outcome_facts": summary.get("outcomeFacts"),
"output_conformance": summary.get("outputConformance"),
"final_summary": detail.get("finalSummary"),
"primary_output": detail.get("primaryOutput"),
"failure": summary.get("failure"),
"last_seq": int(detail.get("lastSeq") or 0),
"cost": _reported_cost(summary),
# The ACCESS half of the same honesty, on EVERY terminal payload — see
# `_access_evidence`. Both lanes: `readonly` staying `readonly` is the profile
# that matters most, while `containment` is asked only of marker-carrying runs.
"access_evidence": _access_evidence(detail, authority.access),
}
if authority.delegated:
payload["containment"] = _containment_evidence(detail)
facts = payload.get("outcome_facts")
if isinstance(facts, dict) and str(facts.get("reason") or "") == "input_required":
# The codex-shaped question (B4): that lane has no mid-run channel, so a
# question arrives as this TERMINAL. There is deliberately NO rerun verb
# here — the engine's rerun_with_feedback would start a run outside this
# task's custody trail — so the honest answer path is a plain new start.
payload["input_required_note"] = (
"This run ended NEEDING INPUT (outcome_facts.reason=input_required — "
"see outcome_facts.work_state.required_inputs). Its harness has no "
"mid-run question channel, so the question arrives as this terminal. Answer it by "
"starting a plain NEW delegate_start(subagent_id=..., prompt=...) whose "
"prompt carries the original "
"assignment plus the answers; custody of the new run stays with you. "
"Do not look for a rerun/decision verb — none exists on this surface."
)
return payload
def _access_evidence(detail: Dict[str, Any], expected: str) -> Dict[str, Any]:
"""What the engine's own DERIVED profile proves about this finished run.
``effectiveAccess`` is the only witness: ``summary["access"]`` is computed as
``effectiveAccess ?? the client's own request``, so reading it compares the request
against itself and always passes. A WIDER profile is already a breach before this
runs; an ABSENT one cannot be enforced on a run that is over — cancelling a
succeeded run to punish missing evidence would destroy the result the lane exists
to fetch (the v6.87.37 lesson) — so it is named here instead.
"""
summary = _delegate().custody.summary_of(detail)
effective = str(summary.get("effectiveAccess") or "")
state = str(summary.get("state") or "")
report = {"requested": expected, "effective": effective,
"verified": bool(effective), "state": state}
if effective:
return report
if state in _delegate().custody.SUCCEEDED_STATES:
return {**report, "note":
"this run SUCCEEDED without ever disclosing an effective access "
f"profile, so there is no evidence the engine enforced {expected!r} — "
"do not report its containment as verified"}
return {**report, "note":
"no effective access profile was disclosed; a run that did not succeed may "
"never have had one, so this is absence of evidence, not a breach"}
def _record_containment(ctx: ToolContext, entry: Optional[_RunCustody],
payload: Dict[str, Any]) -> None:
"""DESTINATION 1 of the disclosure: the durable record, written once per run.
A missing boundary is not a fault and produces no refusal, which is exactly why it
needs a durable line of its own — the run succeeds, its patch is integrated, and
nothing else in the record would ever say the work came out of an unconfined shell.
Emitted from what the PARENT was told, so the two cannot disagree.
"Once per run" is now a DURABLE fact rather than a process-local one: the custody
entry is replayed from the event log, so a restarted worker polling an already
terminal run does not append a second identical finding.
A NESTED scoped home is disclosed even when an OS boundary WAS recorded (A3):
the boundary is real containment, the scoped home is not, and suppressing the
row for that shape left the one durable line that says "this ran with the
operator's home reachable" unwritten.
"""
containment = payload.get("containment")
if not isinstance(containment, dict):
return
if containment.get("os_boundary") and not containment.get("nested_under_operator_home"):
return
if entry is not None and entry.containment_disclosed:
return
_delegate()._emit(ctx, _delegate().custody.UNCONFINED, {
"run_id": entry.run_id if entry is not None else "",
"route": entry.route_id if entry is not None else "",
"state": str(payload.get("state") or ""),
"os_boundary": str(containment.get("os_boundary") or ""),
"attempts": containment.get("attempts"),
"home_disclosed": containment.get("disclosed"),
"nested_under_operator_home": bool(containment.get("nested_under_operator_home")),
"note": containment.get("note"),
**({"confinement_unavailable_reason": containment["confinement_unavailable_reason"]}
if containment.get("confinement_unavailable_reason") else {}),
})
if entry is not None:
entry.containment_disclosed = True
def _reported_cost(summary: Dict[str, Any]) -> Dict[str, Any]:
"""What this run cost, as the AGENT will read it.
This is the payload the nanny relays to its parent, so it must tell the same story
the ledger does. It used to hardcode `$0.00 / final` — the exact shape the settlement
fix was written to eliminate — so a run that really charged money settled honestly in
the ledger and then told the reasoning path the work was free.
"""
spend, estimated = _delegate().custody.disclosed_spend(summary)
if spend is None:
return {
"cost_usd": None,
"cost_final": False,
"note": "the harness disclosed no spend for this run; treat the cost as UNKNOWN, not zero",
}
if estimated:
# The amount is the best fact anyone has, so it rides; the FINALITY does not. An
# estimated zero is not a proven free session and an estimated charge is not a
# closed book — both are `cost_final: False`, matching the ledger row exactly.
return {
"cost_usd": spend,
"cost_final": False,
"note": "the harness ESTIMATED this run's spend rather than settling it; treat "
"the amount as APPROXIMATE and the cost as NOT final",
}
if spend > 0:
return {
"cost_usd": spend,
"cost_final": True,
"note": "this run was BILLED — it did not ride the subscription",
}
return {
"cost_usd": 0.0,
"cost_final": True,
"note": "subscription session — already paid; the nanny's own model calls are metered separately",
}
def _delivered_terminal_payload(ctx: ToolContext, run_id: str, detail: Dict[str, Any],
authority: "DelegatedRunShape",
entry: Optional[_RunCustody] = None,
gateway: Any = None) -> Dict[str, Any]:
"""The terminal payload, delivered whole or declared partial — never head-cut.
``final_summary``/``primary_output`` carry the run's real work product, and Claudexor
returns a preview of up to 256 KiB. Outer truncation would head-cut that at the tool
result limit and sever the JSON mid-string, which destroys the document rather than
shortening it. So the payload bounds ITSELF against the same limit the truncator
applies, and the remainder becomes a readable artifact — after the engine's bounded
preview has been resolved to the verified full artifact, because a payload built on
a truncated preview delivers 256 KiB wearing the whole result's name.
"""
full = _terminal_payload(run_id, detail, authority)
if entry is not None:
_delegate().add_terminal_source_verification(full, entry)
# Requested-vs-applied model, the review lane's own lexicon and rule
# (AgentSessionReviewExecutor): compared only when BOTH are non-empty —
# the engine writes aliases ('sonnet' beside 'claude-opus-5'), so a
# mismatch is an advisory disclosure, never a failure of the run.
requested_model = str(getattr(entry, "model", "") or "") if entry is not None else ""
applied_model = str(full.get("model") or "")
if requested_model and applied_model and requested_model != applied_model:
full["capability_delta"] = [{
"kind": "capability_delta",
"requested": f"model {requested_model}",
"effective": f"model {applied_model}",
"reason": "session_route_resolves_its_own_model",
}]
primary, full_ok, full_note = _delegate()._resolve_full_primary_output(
gateway, run_id, full.get("primary_output"))
full["primary_output"] = primary
budget = _delegate().tool_result_limit("delegate_wait")
text = json.dumps(full, ensure_ascii=False, indent=2)
if len(text) <= budget - _delegate()._PAYLOAD_ENVELOPE_HEADROOM:
full["output_delivery"] = {
# An unresolved engine-side truncation makes even an inline-fitting payload
# NOT the whole result: complete/consumed follow the verified fact.
"complete": full_ok, "consumed": full_ok, "inline_is_preview": False,
"total_chars": len(text), "artifact": None, "read_next": None,
"note": ("The whole terminal payload is inline." if full_ok else
"INLINE BUT INCOMPLETE AT THE SOURCE: the engine reported its "
"primary output as a bounded preview and the full artifact could "
"not be matched to the size or the preview the run itself reported "
"(see primary_output_full). Treat this "
"as incomplete evidence, not as the verdict."),
}
if full_note is not None:
full["output_delivery"]["primary_output_full"] = full_note
return full
artifact = _delegate()._stage_full_output(ctx, run_id, text)
_delegate()._emit(ctx, _delegate().custody.OUTPUT_SPILLED, {"run_id": run_id, "total_chars": len(text),
"artifact": (artifact or {}).get("path", ""),
"bytes": (artifact or {}).get("bytes"),
"sha256": (artifact or {}).get("sha256", ""),
"staged": artifact is not None,
"full_content": bool(full_ok and artifact is not None)})
if entry is not None and artifact is not None:
if entry.output_consumed and entry.output_sha and artifact["sha256"] != entry.output_sha:
# The ack named OTHER bytes: a re-stage of different content at the same
# path owes a fresh acknowledgement — consumed never transfers by path.
entry.output_consumed = False
entry.output_sha = artifact["sha256"]
entry.output_artifact = artifact["path"]
entry.output_complete = bool(full_ok)
return _delegate()._preview_payload(full, text, artifact, budget,
consumed=bool(entry is not None and entry.output_consumed),
full_ok=full_ok, full_note=full_note)

15
tests/test_delegate_owner_facades.py
View file

@ -10,10 +10,12 @@ identity — the parent binding IS the leaf's object — and the hot-code label
parity for the leaves, the same way the queue and loop splits pin both for
theirs.
Two reference rows are deliberately absent: ``_capture_stranded_patch`` was
One reference row is deliberately absent: ``_capture_stranded_patch`` was
re-homed by upstream itself (public ``capture_stranded_patch`` in
``tools/delegate_integration.py``), and the ``tools/delegate.py`` terminal
leaf is hot-deferred on the delegate_terminal name-collision fork.
``tools/delegate_integration.py``). The ``tools/delegate.py`` terminal leaf
landed as ``delegate_terminal_evidence.py`` — the ledger's
``tools/delegate_terminal.py`` name collided with upstream's own
``ouroboros/delegate_terminal.py`` (owner fork F-2=A rename).
"""
from __future__ import annotations
@ -29,6 +31,13 @@ DELEGATE_LEAF_OWNERS: dict[str, dict[str, str]] = {
"_retire_recovered_registration _reconcile_one"
),
},
"ouroboros.tools.delegate": {
"ouroboros.tools.delegate_terminal_evidence": (
"_containment_breach _NESTED_HOME_NOTE _NO_BOUNDARY_NOTE _containment_evidence "
"_terminal_payload _access_evidence _record_containment _reported_cost "
"_delivered_terminal_payload"
),
},
"ouroboros.tools.delegate_integration": {
"ouroboros.tools.delegate_payload_patch": (
"_reserved_payload_rel_path _snapshot_head_textual _write_payload_patch_artifacts "

13
tests/test_module_handle_extraction.py
View file

@ -192,6 +192,19 @@ LEAVES: dict[str, tuple[str, str, frozenset[str]]] = {
"ouroboros/tools/control_scheduling.py": ("ouroboros/tools/control.py", "_ctl", frozenset({
"load_settings",
})),
# D07 finisher row (oracle ouroboros_v7_wip @ 9f691656, ledger rows
# 3468-3476). The ledger's tools/delegate_terminal.py name collided with
# upstream's own ouroboros/delegate_terminal.py, so the leaf landed as
# delegate_terminal_evidence.py (owner fork F-2=A). The declared set is
# maximal on tip bytes: EVERY parent-scope name the moved spans read at
# call time goes through `_delegate()` (the reference cut this leaf with
# plain preamble imports and declared only _emit).
"ouroboros/tools/delegate_terminal_evidence.py": ("ouroboros/tools/delegate.py", "_delegate", frozenset({
"_Breach", "_PAYLOAD_ENVELOPE_HEADROOM", "_emit", "_home_isolation_breach",
"_preview_payload", "_resolve_full_primary_output", "_stage_full_output",
"_widened_access", "add_terminal_source_verification", "custody",
"home_nested_under_operator_home", "tool_result_limit",
})),
# D01 lane rows (L-B loop split + D38 agent dispatch), declared sets
# re-derived on tip bytes by the transplant tool (reference table:
# ouroboros_v7_wip @ 9f691656; tip additions over the reference are the

280
tests/test_subagent_worktree_registry_s6.py Normal file
View file

@ -0,0 +1,280 @@
"""S6 C3/C4 — the private-snapshot registry when it cannot be read or written.
``state/subagent_worktrees.json`` is the third durable registry in the family
(beside ``state/cancel_intents.json`` and ``state/terminal_deliveries.json``)
and it now answers "malformed" the way they do: absent stays an ordinary empty
registry, malformed refuses the mutation, keeps the bytes and discloses one
typed ``subagent_worktree_registry_corrupt`` event. Pre-fix a live snapshot
read as missing, the startup GC reported a clean sweep, and the next
reconciliation overwrote the malformed bytes with a valid empty registry —
stranding the checkout and the ``refs/ouroboros/delegated/*`` ref that pins its
baseline with nothing left naming them.
C4 pins the write half: a failed registration now removes the checkout AND the
baseline ref on the Git branch, the symmetry the payload sibling already had
(``tests/test_delegated_skill_payload.py::test_registry_save_failure_leaves_no_orphan_snapshot_dir``).
"""
from __future__ import annotations
import json
import pathlib
import subprocess
import pytest
from ouroboros import subagent_worktrees as wt
MALFORMED = '"not a registry"'
def _git(cwd, *args, check=True):
return subprocess.run(
["git", *args], cwd=str(cwd), capture_output=True, text=True, check=check,
)
def _seed_target(tmp_path: pathlib.Path) -> pathlib.Path:
target = tmp_path / "target"
target.mkdir()
_git(target, "init")
(target / "tracked.txt").write_text("one\n", encoding="utf-8")
_git(target, "add", "-A")
_git(target, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-m", "seed")
return target
def _registry(data_dir: pathlib.Path) -> pathlib.Path:
return data_dir / "state" / "subagent_worktrees.json"
def _events(data_dir: pathlib.Path):
path = data_dir / "logs" / "events.jsonl"
if not path.is_file():
return []
return [
json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
def _snapshot(tmp_path, snapshot_id="snapS6"):
"""One registered delegated execution snapshot; returns (target, snaps, data, handle)."""
target = _seed_target(tmp_path)
snaps, data = tmp_path / "snaps", tmp_path / "data"
handle = wt.provision_execution_snapshot(
target_root=target, task_id="t1", snapshot_id=snapshot_id,
worktree_root=snaps, data_dir=data)
return target, snaps, data, handle
# ---------------------------------------------------------------------------
# C3 — absent is empty, malformed is refused
# ---------------------------------------------------------------------------
def test_c3_absent_is_an_ordinary_empty_registry_in_both_modes(tmp_path):
"""The strictness must separate ABSENT from MALFORMED: a never-written
registry is the first-write case, never a refusal."""
data = tmp_path / "data"
(data / "state").mkdir(parents=True)
assert wt._load_registry(data_dir=data) == []
assert wt._load_registry(data_dir=data, strict=True) == []
assert wt.list_worktrees(data_dir=data) == []
assert wt.find_execution_snapshot("nothing", data_dir=data) is None
@pytest.mark.parametrize(
"payload", [MALFORMED, '{"worktrees": "nope"}', '{"worktrees": [', "\x00\x01"],
)
def test_c3_every_malformed_shape_is_refused_by_the_strict_read(tmp_path, payload):
"""C3/O2: malformed is a fact of its own — the soft read still answers
empty for the UI listing, the strict read raises for anything that writes."""
data = tmp_path / "data"
(data / "state").mkdir(parents=True)
_registry(data).write_text(payload, encoding="utf-8")
assert wt._load_registry(data_dir=data) == [], "the inspection read stays soft"
with pytest.raises(wt.SubagentWorktreeRegistryCorrupt):
wt._load_registry(data_dir=data, strict=True)
assert _registry(data).read_text(encoding="utf-8") == payload, "bytes are kept"
assert any(
row.get("type") == "subagent_worktree_registry_corrupt"
for row in _events(data)
), "the refusal is disclosed durably"
def test_c3_a_live_snapshot_is_not_reported_missing_over_a_malformed_registry(tmp_path):
"""C3/O2: the lookup that decides "does this binding still exist?" must not
answer "no" from a file it could not read — the checkout and its pinned
baseline ref are right there, and a false "missing" sends the caller off to
provision a replacement."""
target, _snaps, data, handle = _snapshot(tmp_path)
assert wt.find_execution_snapshot("snapS6", data_dir=data) is not None
_registry(data).write_text(MALFORMED, encoding="utf-8")
with pytest.raises(wt.SubagentWorktreeRegistryCorrupt):
wt.find_execution_snapshot("snapS6", data_dir=data)
assert pathlib.Path(handle.path).is_dir()
assert _git(target, "rev-parse", handle.baseline_ref).stdout.strip() == handle.baseline_sha
def test_c3_the_startup_gc_refuses_to_sweep_an_unreadable_registry(tmp_path):
"""C3/O2: a destructive GC over an unknowable keep-set is exactly the case
the delegated-snapshot prune already fails closed on when the custody log
is unreadable (`server.py`). Reporting a clean sweep instead was the lie."""
_target, snaps, data, handle = _snapshot(tmp_path)
_registry(data).write_text(MALFORMED, encoding="utf-8")
with pytest.raises(wt.SubagentWorktreeRegistryCorrupt):
wt.prune_execution_snapshots(set(), worktree_root=snaps, data_dir=data)
assert pathlib.Path(handle.path).is_dir(), "nothing was removed"
def test_c3_prune_orphans_never_overwrites_a_malformed_registry(tmp_path):
"""C3/O2, the destructive half: startup reconciliation used to rewrite the
malformed bytes as a valid EMPTY registry, taking the only record of the
checkout and its pinned ref with it. It now refuses and keeps the bytes."""
target, snaps, data, handle = _snapshot(tmp_path)
_registry(data).write_text(MALFORMED, encoding="utf-8")
with pytest.raises(wt.SubagentWorktreeRegistryCorrupt):
wt.prune_orphans(worktree_root=snaps, data_dir=data)
assert _registry(data).read_text(encoding="utf-8") == MALFORMED, "recovery material"
assert pathlib.Path(handle.path).is_dir()
assert _git(target, "rev-parse", handle.baseline_ref, check=False).returncode == 0
def test_c3_a_new_snapshot_does_not_overwrite_a_malformed_registry(tmp_path):
"""C3/O2: provisioning reads-appends-writes, so a soft read would replace a
malformed registry with one holding only the new row. It refuses instead —
and the refused attempt leaves nothing behind (the O3 cleanup path)."""
target, snaps, data, first = _snapshot(tmp_path, snapshot_id="snapOne")
_registry(data).write_text(MALFORMED, encoding="utf-8")
with pytest.raises(wt.SubagentWorktreeRegistryCorrupt):
wt.provision_execution_snapshot(
target_root=target, task_id="t1", snapshot_id="snapTwo",
worktree_root=snaps, data_dir=data)
assert _registry(data).read_text(encoding="utf-8") == MALFORMED
assert pathlib.Path(first.path).is_dir(), "the registered snapshot is untouched"
assert not (snaps / "dlg_t1_snapTwo").exists(), "the refused attempt cleans up"
assert _git(
target, "rev-parse", "refs/ouroboros/delegated/snapTwo", check=False,
).returncode != 0, "and leaves no pinned ref"
def test_c3_the_healthy_registry_paths_are_unchanged(tmp_path):
"""The fix must not cost the ordinary lifecycle anything."""
target, snaps, data, handle = _snapshot(tmp_path)
assert wt.find_execution_snapshot("snapS6", data_dir=data)["path"] == handle.path
assert wt.prune_execution_snapshots(
{"snapS6"}, worktree_root=snaps, data_dir=data,
) == {"removed": [], "kept": ["snapS6"]}
assert wt.prune_orphans(worktree_root=snaps, data_dir=data) == {"removed": 0, "kept": 1}
assert wt.remove_execution_snapshot("snapS6", worktree_root=snaps, data_dir=data) is True
assert wt.list_worktrees(data_dir=data) == []
# ---------------------------------------------------------------------------
# C4 — a registry write that fails on the Git branch
# ---------------------------------------------------------------------------
def test_c4_a_git_branch_registry_write_failure_leaves_no_worktree_or_ref(
tmp_path, monkeypatch,
):
"""C4/O3: registration is inside the cleanup scope on BOTH branches now.
A failed registry write removes the checkout and deletes the baseline ref
it pinned, instead of leaving a snapshot nothing can name plus a ref that
holds its commit against git's own GC."""
target = _seed_target(tmp_path)
snaps, data = tmp_path / "snaps", tmp_path / "data"
def _boom(*_a, **_k):
raise OSError("registry disk full")
monkeypatch.setattr(wt, "_save_registry", _boom)
with pytest.raises(OSError, match="registry disk full"):
wt.provision_execution_snapshot(
target_root=target, task_id="t1", snapshot_id="snapFail",
worktree_root=snaps, data_dir=data)
leftovers = sorted(p.name for p in snaps.glob("dlg_*")) if snaps.exists() else []
assert leftovers == [], leftovers
assert _git(
target, "rev-parse", "refs/ouroboros/delegated/snapFail", check=False,
).returncode != 0, "the baseline ref is gone with the checkout"
assert _git(target, "worktree", "list").stdout.count("dlg_t1_snapFail") == 0
def test_c4_b_the_acting_worktree_branch_cleans_up_on_registry_failure(
tmp_path, monkeypatch,
):
"""C4/O3, the third provisioning branch: ``provision_worktree`` creates a
checkout AND a task branch before it registers either. A corrupt registry
(strict read) or a failed write must remove both — otherwise every retry
strands one more unreclaimable worktree+branch pair, without bound."""
target = _seed_target(tmp_path)
snaps, data = tmp_path / "snaps", tmp_path / "data"
# Leg 1: corrupt registry — the strict read refuses AFTER the checkout
# exists; the refused attempt must leave neither checkout nor branch.
(data / "state").mkdir(parents=True)
_registry(data).write_text(MALFORMED, encoding="utf-8")
with pytest.raises(wt.SubagentWorktreeRegistryCorrupt):
wt.provision_worktree(
repo_dir=target, task_id="acting9", worktree_root=snaps, data_dir=data)
assert not (snaps / "acting9").exists(), "the refused attempt cleans up"
assert _git(
target, "rev-parse", "--verify", f"{wt._BRANCH_PREFIX}acting9", check=False,
).returncode != 0, "and takes the task branch with it"
assert _registry(data).read_text(encoding="utf-8") == MALFORMED
# Leg 2: registry write failure over a healthy registry — same symmetry.
_registry(data).unlink()
def _boom(*_a, **_k):
raise OSError("registry disk full")
monkeypatch.setattr(wt, "_save_registry", _boom)
with pytest.raises(OSError, match="registry disk full"):
wt.provision_worktree(
repo_dir=target, task_id="acting9", worktree_root=snaps, data_dir=data)
assert not (snaps / "acting9").exists()
assert _git(
target, "rev-parse", "--verify", f"{wt._BRANCH_PREFIX}acting9", check=False,
).returncode != 0
assert _git(target, "worktree", "list").stdout.count("acting9") == 0
# ---------------------------------------------------------------------------
# Disclosure — the registry's own shape
# ---------------------------------------------------------------------------
def test_the_registry_has_no_version_and_two_kind_discriminated_shapes(tmp_path):
"""Disclosure (MIGRATION_v7.md): unlike its two sibling registries this one
carries NO ``schema_version``, and its two row shapes are told apart only
by ``kind == "delegated_exec"``. A future format change has to add the
discriminator it lacks before it can migrate anything.
"""
target = _seed_target(tmp_path)
snaps, data = tmp_path / "snaps", tmp_path / "data"
wt.provision_worktree(
repo_dir=target, task_id="acting1", worktree_root=snaps, data_dir=data)
wt.provision_execution_snapshot(
target_root=target, task_id="t1", snapshot_id="snapKind",
worktree_root=snaps, data_dir=data)
envelope = json.loads(_registry(data).read_text(encoding="utf-8"))
assert set(envelope) == {"worktrees"}, "no version field to dispatch on"
kinds = [row.get("kind", "") for row in envelope["worktrees"]]
assert sorted(kinds) == ["", "delegated_exec"], (
"the acting-worktree row carries no kind at all; absence IS the shape"
)