Merge pull request #926: preserve model recovery and failure evidence

Preserve supervised-task recovery, private response-failure evidence and delegated snapshot bytes. Include the reviewed Windows CI repairs.

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

.github/workflows/ci.yml

@@ -224,6 +224,19 @@ jobs:
       - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4
         with:
           node-version: '22'
+      - name: Set up Windows POSIX test utilities
+        if: runner.os == 'Windows'
+        id: posix-tools
+        uses: msys2/setup-msys2@66cd2cce69caa17b53920067426061ca1de3a884 # v2
+        with:
+          msystem: MSYS
+          release: false
+          update: true
+          install: rsync
+      - name: Select Windows POSIX test utilities
+        if: runner.os == 'Windows'
+        shell: pwsh
+        run: '"${{ steps.posix-tools.outputs.msys2-location }}\usr\bin" >> $env:GITHUB_PATH'
       # The hermetic commit gate runs this exact suite (ouroboros/preflight_node.py);
       # `shell: bash` keeps the glob working on the Windows runner too.
       - name: Run browser-module tests (node --test — mirrored by the hermetic commit gate)

docs/DEVELOPMENT.md

@@ -1944,6 +1944,10 @@ both critical. The imperatives:
   invariant and the tool description; every other guard (owner terminality,
   top-level principal, proven drift, protected paths, staged-never-committed)
   is unchanged (`tests/test_delegated_run_isolation_orphans.py`).
+  Snapshot tests compare raw LF/CRLF inputs under Git checkout filters, require
+  zero patch before child edits, and retain normal Git apply semantics afterwards.
+  Copy failures or a source change against the baseline leave no registered
+  snapshot or pinned ref (`tests/test_snapshot_file_inputs.py`).
 - Outcome honesty: a delegating parent must not produce a clean no-tool
   final answer while direct children run undecided — one bounded absorption
   reminder, then best-effort (`children_unabsorbed`); while that gate is
@@ -2210,7 +2214,14 @@ Focused regressions: `test_review_late_cas_recovery.py`, `test_delivery_control_
   operation; never inject its credentials, run its tools, compact inside the
   adapter or silently repeat a generation. Recover a local connection loss using
   the same operation ID; record unknown outcomes as unknown. ACK only after the
-  existing private CAS owns the exact result. Optional host hints must be chosen
+  existing private CAS owns the exact result. Failed-response capture uses the
+  operation catalog's optional query, frozen before create and reused with the
+  same idempotency key; absence preserves the strict legacy result shape. Keep
+  full received bytes/exception chains private and compact diagnostics in the
+  ordinary problem context. A known terminal with unusable output is a settled
+  provider result plus local rejection, never unknown or not-dispatched: preserve
+  both existing stream-rejection markers across sync, async and process boundaries.
+  Local rejection must not rotate accounts. Optional host hints must be chosen
   by their caller according to transport capability; explicit unsupported options
   refuse, rather than being silently removed and retried. Record submitted model
   options beside the engine's applied options on the usage row; an absent report
@@ -2428,8 +2439,11 @@ by "Provider Independence" above. Call-site imperatives:
   on the wire before a later reservation can be refused), so the record keeps
   the attempt booked and the budget terminal, not the provider terminal, ends
   the round; the bounded repeat rail belongs to interactive primary rounds. Ordinary managed
-  tasks and native API children use upstream-observed continuation, while exact
-  session nanny routes keep their independent hold. Every other caller — forced-final, fallback
+  tasks and native API children use upstream-observed continuation. Exact
+  session supervisors first try the live-leaf hold and otherwise use ordinary
+  managed recovery for their own model; no consumed/terminal/patch-disposition
+  predicate gates cognition. A successful hold closes and clears any prior
+  transport episode, so its acknowledged wake alone resumes the model. Every other caller — forced-final, fallback
   candidates, review actors, safety, external-harness delegated runs — keeps
   `transport_death_retries=0`. A round that holds a transport-death repeat
   record sends nothing further except the typed-death repeats — a repeat that

ouroboros/delegate_hold.py

@@ -18,10 +18,11 @@ Eligibility is deliberately narrow (owner Q8=A): configured-session/exact
 actor routes with EXACTLY one open, non-terminal delegated run and no pending
 invocations. The liveness probe is a READ-ONLY engine poll requiring a
 positive non-terminal engine state — a refusal, fault, empty state, or open
-containment fault is not evidence of a live leaf and takes today's terminal
-path. A terminal-but-unsettled leaf also takes today's terminal path, whose
-completion-wins reconciliation already preserves the leaf's output — holding
-there would spin an instant-wake paid loop.
+containment fault is not evidence of a live leaf. When no hold applies, the
+supervising model uses ordinary managed transport recovery; external custody
+remains unchanged. A terminal-but-unsettled leaf must not hold, since an
+instant terminal wake would buy repeated model calls. A successful hold takes
+over any existing transport episode, so its wake alone permits the next round.
 """
 
 from __future__ import annotations
@@ -151,12 +152,12 @@ def _unknown_attempt_id() -> str:
 
 def latch_after_unknown(
     tools: Any, *, error_kind: str, drive_logs: pathlib.Path, task_id: str,
-    emit_progress: Any,
+    emit_progress: Any, transport_episode: Any = None,
 ) -> bool:
-    """Round-gate entry: latch a hold instead of terminalizing.
+    """Round-gate entry: prefer a live leaf to a new model recovery episode.
 
     Returns True when the caller should ``continue`` to the next round top,
-    where the durable latch parks the task in ``hold_step``.
+    clearing its transport episode; the durable latch parks in ``hold_step``.
     """
     if str(error_kind or "") != "provider_outcome_unknown":
         return False
@@ -178,6 +179,15 @@ def latch_after_unknown(
         **({"unknown_attempt_id": attempt_id} if attempt_id else {}),
     }
     write_unknown_hold(ctx, run_id, hold)
+    if transport_episode is not None:
+        from ouroboros.loop_transport import emit_network_wait_event
+
+        emit_network_wait_event(
+            drive_logs, task_id=task_id, phase="ended",
+            elapsed_sec=transport_episode.waited_sec, redials=transport_episode.redials,
+            model=str(getattr(ctx, "active_model", "") or ""), detail="hold_latched",
+            outcome_custody=transport_episode.outcome_custody,
+        )
     _emit_hold_event(
         drive_logs, task_id=task_id, phase="entered", run_id=run_id,
         hold_cycles=hold["hold_cycles"], attempt_id=attempt_id,
@@ -300,6 +310,7 @@ def hold_step(
         return "terminal"
 
     if controls.get("finalize_now"):
+        ctx._transport_repeat_control_reason = str(controls["finalize_now"]).splitlines()[0].strip()
         return _exit_terminal("finalize_now")
     if new_input:
         # The round-top drain already appended owner/task dialogue: that IS the
@@ -336,6 +347,9 @@ def hold_step(
         pend_payload = None
     if (_wake_is_control(payload) or _control_wakes(ctx)
             or (isinstance(pend_payload, dict) and _wake_is_control(pend_payload))):
+        from ouroboros.loop_transport import transport_repeat_stop_requested
+
+        transport_repeat_stop_requested(ctx)  # Preserve a current owner cause without delivering/ACKing it.
         return _exit_terminal("control_wake")
     if str(payload.get("status") or "") in _NON_WAKE_STATUSES:
         # A refusal/fault is a daemon statement, not a leaf wake: no proof of a

ouroboros/gateways/claudexor.py

@@ -286,17 +286,28 @@ def _is_loopback(host: str) -> bool:
         return False
 
 
-def account_catalog_supported(operations: list[dict], path: str) -> bool:
-    """Opt in only when this exact operation declares the accounts query view."""
+def operation_query_supported(operations: list[dict], *, method: str, path: str,
+                              name: str, value: str) -> bool:
+    """Negotiate an exact query value from the serving operation's descriptor."""
     return any(
-        operation.get("method") == "GET" and operation.get("path") == path
-        and any(parameter.get("name") == "view" and parameter.get("location") == "query"
-                and "accounts" in (parameter.get("enum") or [])
-                for parameter in operation.get("parameters", []) if isinstance(parameter, dict))
+        operation.get("method") == method and operation.get("path") == path
+        and any(parameter.get("name") == name and parameter.get("location") == "query"
+                and isinstance(parameter.get("enum"), list) and value in parameter["enum"]
+                for parameter in (operation.get("parameters") or []) if isinstance(parameter, dict))
         for operation in operations if isinstance(operation, dict)
     )
 
 
+def account_catalog_supported(operations: list[dict], path: str) -> bool:
+    """Opt in only when this exact operation declares the accounts query view."""
+    return operation_query_supported(operations, method="GET", path=path, name="view", value="accounts")
+
+
+def model_failure_evidence_supported(operations: list[dict]) -> bool:
+    return operation_query_supported(operations, method="POST", path="/v2/model-operations",
+                                     name="captureFailureEvidence", value="true")
+
+
 class ClaudexorGateway:
     """Thin typed client over the Claudexor ``/v2`` control API."""
 
@@ -568,11 +579,12 @@ class ClaudexorGateway:
         return ref
 
     def create_model_operation(self, request_ref: Dict[str, Any], *,
-                               idempotency_key: str) -> Dict[str, Any]:
+                               idempotency_key: str, capture_failure_evidence: bool = False) -> Dict[str, Any]:
         """Create or rejoin exactly one caller-identified generation; never mint a retry key."""
         key = _model_idempotency_key(idempotency_key)
+        path = "/v2/model-operations" + ("?captureFailureEvidence=true" if capture_failure_evidence else "")
         return _model_operation(self._request(
-            "POST", "/v2/model-operations", json_body={"request": _model_payload_ref(request_ref)},
+            "POST", path, json_body={"request": _model_payload_ref(request_ref)},
             headers={"Idempotency-Key": key},
         ))
 

ouroboros/llm_claudexor.py

@@ -54,7 +54,9 @@ from ouroboros._usage_response import provider_cost_value
 from ouroboros.anthropic_native_custody import scrub_native_custody
 from ouroboros.claudexor_daemon import ensure_owned_gateway, owned_engine_version, read_owned_gateway
 from ouroboros.deadline_utils import llm_transport_timeout_sec
-from ouroboros.gateways.claudexor import ClaudexorUnavailable, engine_at_least, _READ_TIMEOUT_SEC
+from ouroboros.gateways.claudexor import (
+    ClaudexorUnavailable, engine_at_least, model_failure_evidence_supported, _READ_TIMEOUT_SEC,
+)
 from ouroboros.llm_attempt import _attempt_request, _candidate_before_dispatch
 from ouroboros.model_slots import MODEL_ACCOUNTS_KEY, model_role_option
 from ouroboros.model_wait import ModelWaitInterrupted, current_model_wait, prepared_call_scope
@@ -148,6 +150,11 @@ class ClaudexorModelError(RuntimeError):
         self.status_code = 0 if unknown else int(context.get("httpStatus") or 0)
         self.reset_at = str(context.get("resetsAt") or "")
         self.retryable = False if unknown else problem.get("retryable") is True
+        if self.code == "response_rejected":
+            # Reconstructed process-boundary errors retain the same local
+            # rejection: no unknown outcome or same-request wire repair.
+            self.stream_rejected = self.stream_incomplete = True
+            self.retryable = False
         self.model_role = model_role
         self.operation_id = operation_id
         self.route = copy.deepcopy(route or {})
@@ -156,9 +163,12 @@ class ClaudexorModelError(RuntimeError):
     def display_message(self) -> str:
         """Show typed provider details without changing exception classification text."""
         context = self.problem.get("context") or {}
-        details = [] if self.code == "model_outcome_unknown" else [
+        fields = (("stage", "stage"), ("errorCode", "cause"))
+        if self.code != "model_outcome_unknown":
+            fields += (("vendorCode", "provider_code"), ("parameter", "parameter"))
+        details = [
             f"{label}={value.strip()}"
-            for key, label in (("vendorCode", "provider_code"), ("parameter", "parameter"))
+            for key, label in fields
             if isinstance(value := context.get(key), str) and value.strip()
         ]
         # Details lead so the existing terminal preview can name the refusal.
@@ -280,6 +290,8 @@ def adopt_turn_state(slot: ModelTurnState | None, payload: dict, result: dict) -
 
 
 def _remember_failed_profile(target: dict, parameters: dict, error: ClaudexorModelError) -> None:
+    if getattr(error, "stream_rejected", False):
+        return  # Local message normalization says nothing about account readiness.
     route = error.route or {}
     key = (parameters.get("cache_affinity"), route.get("source"), route.get("model"))
     if (key == (parameters.get("cache_affinity"), target["source"], target["resolved_model"])
@@ -391,6 +403,7 @@ class _ModelInvocation:
         self.request_manifest_ref: dict = {}
         self.interrupt_reason = ""
         self.create_attempted = False
+        self.capture_failure_evidence = False
         self.defer_close = False
         self.io_active = False
         self.io_lock = threading.Lock()
@@ -429,11 +442,15 @@ class _ModelInvocation:
         self.check_control()
         try:
             self.gateway = ensure_owned_gateway()
+            # Freeze once before create. A lost create reply or replaced gateway
+            # must reuse this same operation's diagnostic/idempotency contract.
+            self.capture_failure_evidence = model_failure_evidence_supported(self.gateway.operations())
             self.request_ref = self.gateway.upload_model_request(self.payload, idempotency_key=self.invocation_id)
             self.request_manifest_ref = persist_call(self.root, task_id=self.task_id, call_id=f"{self.invocation_id}_model_request",
                          call_type="llm_claudexor_request", payload=self.payload, keep_raw=True,
                          manifest={"invocation_id": self.invocation_id, "request_ref": self.request_ref,
-                                   "model_role": self.role})["manifest_ref"]
+                                   "model_role": self.role,
+                                   "capture_failure_evidence": self.capture_failure_evidence})["manifest_ref"]
         except ClaudexorUnavailable as error:
             raise ClaudexorModelError({"code": error.code, "message": str(error)}, model_role=self.role) from None
 
@@ -447,7 +464,8 @@ class _ModelInvocation:
                     self.root, task_id=self.task_id, call_id=f"{self.invocation_id}_model_request",
                     call_type="llm_claudexor_request", payload=self.payload, keep_raw=True,
                     manifest={"invocation_id": self.invocation_id, "request_ref": self.request_ref,
-                              "model_role": self.role, "operation_id": self.operation_id})["manifest_ref"]
+                              "model_role": self.role, "operation_id": self.operation_id,
+                              "capture_failure_evidence": self.capture_failure_evidence})["manifest_ref"]
             except Exception as error:
                 self.request_manifest_ref = {}
                 log.warning("Model custody checkpoint unavailable: %s", type(error).__name__)
@@ -475,7 +493,8 @@ class _ModelInvocation:
                 if not self.operation_id:
                     self.create_attempted = True
                     self.observe_operation()
-                    detail = self.gateway.create_model_operation(self.request_ref, idempotency_key=self.invocation_id)
+                    detail = self.gateway.create_model_operation(self.request_ref, idempotency_key=self.invocation_id,
+                        **({"capture_failure_evidence": True} if self.capture_failure_evidence else {}))
                     self.operation_id = detail["id"]
                     self.observe_operation(accepted=True)
                 else:
@@ -664,7 +683,10 @@ class _ModelInvocation:
             raise error
         message = result.get("message")
         if not isinstance(message, dict):
-            error = self.error({"code": "malformed_response", "message": "The provider returned no model message."}, unknown=True)
+            error = ClaudexorModelError(result.get("problem") or {
+                "code": "response_rejected", "message": "The terminal provider response contained no usable model message."},
+                model_role=self.role, operation_id=self.operation_id, route=route)
+            error.stream_rejected = error.stream_incomplete = True
             error.physical_attempt_capture = self.capture
             error.usage = usage
             raise error

ouroboros/loop.py

@@ -576,8 +576,12 @@ def run_llm_loop(
                 active_model, active_use_local = model_call.active_model, model_call.active_use_local
                 context_fit_plan = model_call.context_fit_plan
             tools._ctx._current_llm_call_meta = dict(accumulated_usage.get("_last_llm_call_meta") or {})
-
             last_error_kind = str(accumulated_usage.get("_last_llm_error_kind") or "")
+            if msg is None and _delegate_hold_latch(
+                    tools, error_kind=last_error_kind, drive_logs=drive_logs,
+                    task_id=task_id, emit_progress=emit_progress, transport_episode=transport_wait):
+                transport_wait = None  # The leaf wake owns resumption, without a provider probe.
+                continue
             transport_wait = _reconcile_transport_wait(
                 transport_wait, ctx, msg_present=msg is not None, error_kind=last_error_kind,
                 drive_logs=drive_logs, task_id=task_id, model=active_model, emit_progress=emit_progress)
@@ -614,10 +618,6 @@ def run_llm_loop(
                 emit_progress=emit_progress, incoming_messages=incoming_messages, owner_msg_seen=_owner_msg_seen):
                 free_redial = True
                 continue
-            if msg is None and _delegate_hold_latch(
-                    tools, error_kind=last_error_kind, drive_logs=drive_logs,
-                    task_id=task_id, emit_progress=emit_progress):  # hold latched -> next round top parks
-                continue
             if msg is None:
                 # Exact actor routes skip generic substitution and fail as infrastructure.
                 text, accumulated_usage, forced_trace = _handle_provider_unavailable(

ouroboros/loop_transport.py

@@ -143,10 +143,9 @@ def emit_network_wait_event(
 
 
 def managed_transport_continuation(ctx: Any) -> bool:
-    """Owner-selected continuation applies to ordinary managed cognition."""
+    """Managed cognition may recover; a live delegated-leaf hold takes priority."""
     return bool(ctx is not None and getattr(ctx, "task_id", "")
-                and not getattr(ctx, "is_direct_chat", False)
-                and getattr(ctx, "_configured_subagent_route_kind", "") != "agent_session")
+                and not getattr(ctx, "is_direct_chat", False))
 
 
 def continue_unknown_transport(episode: TransportWaitEpisode, *, llm: Any, tools: Any,

ouroboros/subagent_worktrees.py

@@ -477,7 +477,9 @@ def provision_execution_snapshot(
 
     Baseline construction never touches the target's own index, HEAD or working
     files: a TEMPORARY index is seeded from HEAD and stages the eligible
-    tracked/staged/text inputs (``.gitignore`` respected). Binary and large
+    tracked/staged/text inputs (``.gitignore`` respected). The execution copy
+    retains original regular-file bytes despite Git checkout filters; a Git
+    comparison binds the copied content to that same baseline. Binary and large
     untracked inputs are streamed into the execution root outside Git's ODB,
     with exact preimages retained in the existing snapshot record. The Git tree
     is committed as a synthetic baseline pinned by a ref under
@@ -583,6 +585,24 @@ def provision_execution_snapshot(
         try:
             wt_path.parent.mkdir(parents=True, exist_ok=True)
             _git(target, "worktree", "add", "--detach", str(wt_path), baseline_sha)
+            from ouroboros.artifacts import copy_artifact_file
+
+            # Git owns the baseline representation, but the child must see the
+            # source's actual working bytes, not checkout's CRLF/smudge rewrite.
+            # Read the existing tree inventory so deletions, links and gitlinks
+            # keep Git's semantics and excluded paths can never enter the copy.
+            for item in manifest_raw.split(b"\0"):
+                metadata, separator, raw_path = item.partition(b"\t")
+                if separator and metadata.split()[0] in (b"100644", b"100755"):
+                    relative = raw_path.decode("utf-8", errors="surrogateescape")
+                    original = target / relative
+                    if original.is_symlink():
+                        raise OSError(f"snapshot input changed from a regular file: {relative}")
+                    copy_artifact_file(original, wt_path / relative)
+            # A concurrent source edit must not appear as the child's work.
+            # Use the same Git representation as ordinary patch capture, once
+            # for the whole tree, before the separately tracked file inputs.
+            _git(wt_path, "diff", "--quiet", "--no-ext-diff", baseline_sha, "--")
             from ouroboros.workspace_file_outputs import copy_snapshot_file_inputs
             file_baseline = copy_snapshot_file_inputs(target, wt_path, file_inputs)
             if file_baseline:

tests/test_agency_direct_targets.py

@@ -1,6 +1,7 @@
 """Certain utility writes use resource authority without interpreting program bodies."""
 from __future__ import annotations
 
+import os
 import pathlib
 import shlex
 import sys
@@ -189,7 +190,9 @@ def test_explicit_utility_roles_preserve_source_and_destination(resources, monke
         "tar_archive_first": ["tar", "-xf", str(archive), "-C", str(target)],
         "tar_directory_first": ["tar", "-C", str(target), "-xf", str(archive)],
         "gzip": ["gzip", str(output)],
-        "rsync": ["rsync", str(source), str(output)],
+        # A Windows drive colon selects an rsync remote host; these operands are local.
+        "rsync": ["rsync", pathlib.Path(os.path.relpath(source, workspace)).as_posix(),
+                  pathlib.Path(os.path.relpath(output, workspace)).as_posix()],
     }
     declared = target if utility.startswith("tar_") else pathlib.Path(str(output) + ".gz") if utility == "gzip" else output
     result = registry.execute_result("run_command", {"cmd": commands[utility], "outputs": [str(declared)]})

tests/test_agency_execution.py

@@ -87,7 +87,10 @@ def test_safety_keeps_full_task_sources_after_transcript_compaction(tmp_path, mo
     prompt = safety._build_check_prompt("run_command", {"cmd": ["gh", "repo", "create"]},
         [{"role": "user", "content": "compacted conversation"}], ctx=ctx, resolved_binding=binding)
     assert origin in prompt and reply in prompt
-    assert "answer-1" in prompt and str(work) in prompt
+    assert "answer-1" in prompt
+    source_json = prompt.split("Task sources and physical target (complete; provenance is not consent):\n", 1)[1]
+    facts, _ = json.JSONDecoder().raw_decode(source_json)
+    assert facts["resolved_target"]["target_path"] == repr(work)
     assert "Create the requested private repository" in prompt
 
 

tests/test_agency_file_policy.py

@@ -120,7 +120,7 @@ def test_declared_outputs_are_created_once_and_captured_with_exact_bytes(files):
     registry, ctx, _home, _work, data = files
     payloads = {"report.log": "Log deliverable\n", "build.manifest": "Manifest deliverable\n", "tokens.json": '{"count":7}\n'}
     body = "from pathlib import Path\n"
-    body += "for name, text in " + repr(payloads) + ".items():\n Path(name).write_text(text, encoding='utf-8')\n"
+    body += "for name, text in " + repr(payloads) + ".items():\n Path(name).write_bytes(text.encode('utf-8'))\n"
     body += "with Path('executions.txt').open('a') as out: out.write('once\\n')\n"
     result = registry.execute_result("run_command", {"cmd": [sys.executable, "-c", body],
         "cwd": "task_drive", "outputs": list(payloads)})
@@ -145,7 +145,7 @@ def test_workspace_patch_keeps_log_manifest_and_token_report(files, tmp_path):
         subprocess.run(["git", *args], cwd=work, check=True, capture_output=True)
     payloads = {"report.log": "LOG_PAYLOAD\n", "build.manifest": "MANIFEST_PAYLOAD\n", "tokens.json": '{"report":7}\n'}
     for name, content in payloads.items():
-        (work / name).write_text(content, encoding="utf-8")
+        (work / name).write_bytes(content.encode("utf-8"))
     output = tmp_path / "captured"
     _records, manifest = write_workspace_patch_artifacts(work, output, task={})
     assert set(payloads) <= set(manifest["untracked_included"]), manifest

tests/test_agency_model_wait.py

@@ -58,7 +58,7 @@ def test_resource_refusal_does_not_suspend_cyber_execution(live_wait, tmp_path,
         events = [json.loads(line) for line in (root / "logs/events.jsonl").read_text().splitlines()]
         advisory = next(row for row in events if row["type"] == "safety_advisory")
         assert advisory["assessment_allowed"] is False and code in advisory["assessment"]
-    assert len(transport.operations) == 1
+    assert len(transport.accepted_operations) == 1
     assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "released"]
     assert current_model_wait() is controller and not controller.closed
     assert "wait_for_resources" not in json.dumps(transport.uploads[0][0])

tests/test_agency_protected_operands.py

@@ -1,6 +1,7 @@
 """Protected operands follow real file roles and independent shell redirections."""
 from __future__ import annotations
 
+import os
 import pathlib
 import shlex
 import sys
@@ -95,8 +96,11 @@ def test_execute_output_capture_and_ordinary_input_remain_usable(protected):
     assert result.status == "ok" and (workspace / "copy.txt").read_bytes() == (workspace / "driver.txt").read_bytes(), result.text
 
 
-def test_literal_argv_operators_globs_and_descriptors_keep_their_roles(protected):
+def test_literal_argv_operators_globs_and_descriptors_keep_their_roles(protected, monkeypatch):
     registry, _ctx, reference, _workspace = protected
+    # Native Windows parents otherwise trigger MSYS/Cygwin globbing before rm sees argv.
+    for name in ("MSYS", "CYGWIN"):
+        monkeypatch.setenv(name, f"{os.environ.get(name, '')} noglob".strip())
     original = reference.read_bytes()
     for cmd in (["echo", "<", "reference"], ["echo", ">", "reference"], ["rm", "-f", "ref*"]):
         result = registry.execute_result("run_command", {"cmd": cmd})

tests/test_claudexor_failure_evidence.py

@@ -0,0 +1,198 @@
+"""Complete failed model evidence and known-terminal rejection share existing custody."""
+
+import asyncio
+import base64
+from copy import deepcopy
+from dataclasses import asdict
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from ouroboros import llm_claudexor as transport, usage_accounting as ua
+from ouroboros.gateways.claudexor import ClaudexorUnavailable
+from ouroboros.loop_llm_call import classify_llm_exception
+from ouroboros.request_wire_recovery import plan_next_wire_retry
+from ouroboros.tools import vision_process
+from ouroboros.transport_custody import ProviderNotDispatched
+from tests.test_llm_claudexor import Gateway, MODEL, ROUTE, ledger, result, retained, setup as setup
+
+
+CAPTURE_OPERATION = {"method": "POST", "path": "/v2/model-operations", "parameters": [
+    {"name": "captureFailureEvidence", "location": "query", "enum": ["true", "false"]}]}
+REJECTION = {"code": "response_rejected", "message": "The terminal response could not form a model message.",
+             "retryable": False, "context": {"stage": "message", "requestId": "request-one"}}
+
+
+def failure_evidence(body=b'\xffprivate-wire-marker\r\ndata: invalid JSON\n\n'):
+    return {"bodyBase64": base64.b64encode(body).decode("ascii"), "receivedBytes": len(body),
+            "bodyComplete": False, "stage": "message", "causeCycle": False,
+            "errors": [{"name": "SyntaxError", "message": "private-error-marker", "stack": "private-stack-marker",
+                        "code": None}]}
+
+
+def call(client, asynchronous, **kwargs):
+    value = (client.chat_async if asynchronous else client.chat)([], MODEL, **kwargs)
+    return asyncio.run(value) if asynchronous else value
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("supported", [False, True])
+def test_capture_is_negotiated_once_without_changing_provider_payload(setup, asynchronous, supported):
+    root, gateway, client = setup
+    gateway.operation_catalog = [deepcopy(CAPTURE_OPERATION)] if supported else []
+    call(client, asynchronous, model_role="main")
+    assert gateway.catalog_reads == 1
+    assert gateway.capture_requests == ([{"capture_failure_evidence": True}] if supported else [{}])
+    payload = gateway.uploads[0][0]
+    assert "captureFailureEvidence" not in json.dumps(payload)
+    assert "capture_failure_evidence" not in json.dumps(payload)
+    assert retained(root, "request") == payload
+    manifests = list((root / "observability/calls/task-one").glob("*_model_request.json"))
+    assert len(manifests) == 1
+    manifest = json.loads(manifests[0].read_text())
+    assert manifest["capture_failure_evidence"] is supported
+    assert manifest["operation_id"] == "op-0"
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_lost_create_and_gateway_replacement_reuse_frozen_capture(setup, monkeypatch, asynchronous):
+    root, gateway, client = setup
+    gateway.operation_catalog = [deepcopy(CAPTURE_OPERATION)]
+    gateway.lose_create = True
+    replacement = Gateway()
+    for name in ("accepted_operations", "creates", "capture_requests"):
+        setattr(replacement, name, getattr(gateway, name))
+    monkeypatch.setattr(transport, "read_owned_gateway", lambda: replacement)
+    monkeypatch.setattr(transport, "current_model_wait", lambda: SimpleNamespace(
+        tool_context=SimpleNamespace(task_id="task-one", is_direct_chat=False), control_reason=lambda: None))
+    monkeypatch.setattr(transport.config, "NETWORK_WAIT_BACKOFF_START_SEC", 0.001)
+    monkeypatch.setattr(transport.config, "NETWORK_WAIT_BACKOFF_MAX_SEC", 0.001)
+    answer, usage = call(client, asynchronous)
+    assert answer == result()["message"]
+    assert gateway.catalog_reads == 1 and replacement.catalog_reads == 0
+    assert gateway.capture_requests == [{"capture_failure_evidence": True}] * 2
+    assert len(gateway.creates) == 2 and len(set(gateway.creates)) == 1
+    assert len(gateway.accepted_operations) == len(usage["ledger_attempt_ids"]) == 1
+    assert gateway.closed == replacement.closed == 1
+    assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "settled"]
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+def test_catalog_read_failure_does_not_guess_unsupported_or_dispatch(setup, monkeypatch, asynchronous):
+    root, gateway, client = setup
+
+    def failed_catalog():
+        raise ClaudexorUnavailable("daemon_unreachable", "Metadata connection lost")
+
+    monkeypatch.setattr(gateway, "operations", failed_catalog)
+    with pytest.raises(transport.ClaudexorModelError) as caught:
+        call(client, asynchronous)
+    assert caught.value.code == "daemon_unreachable"
+    assert caught.value.physical_attempt_capture.state == "released"
+    assert not gateway.uploads and not gateway.creates and gateway.closed == 1
+    assert [row["state"] for row in ledger(root)] == ["reserved", "released"]
+
+
+@pytest.mark.parametrize("asynchronous", [False, True])
+@pytest.mark.parametrize("outcome", ["completed", "incomplete"])
+@pytest.mark.parametrize("has_problem", [False, True])
+def test_known_terminal_null_message_settles_then_rejects_without_private_projection(
+    setup, caplog, asynchronous, outcome, has_problem,
+):
+    root, gateway, client = setup
+    gateway.operation_catalog = [deepcopy(CAPTURE_OPERATION)]
+    evidence = failure_evidence()
+    original_problem = deepcopy(REJECTION) if has_problem else None
+    gateway.results = [{**result(outcome=outcome, problem=original_problem),
+                        "message": None, "failureEvidence": evidence}]
+    acknowledge = gateway.acknowledge_model_result
+
+    def retained_first(*args):
+        assert retained(root) == gateway.results[0]
+        assert ledger(root)[-1]["state"] == "settled"
+        return acknowledge(*args)
+
+    gateway.acknowledge_model_result = retained_first
+    with pytest.raises(transport.ClaudexorModelError) as caught:
+        call(client, asynchronous, model_role="vision")
+    error = caught.value
+    assert type(error) is transport.ClaudexorModelError
+    assert not isinstance(error, ProviderNotDispatched)
+    assert error.code == "response_rejected" and error.stream_rejected and error.stream_incomplete
+    assert error.operation_id == "op-0" and error.model_role == "vision" and error.route == ROUTE
+    if has_problem:
+        assert error.problem == original_problem
+    assert error.physical_attempt_capture.state == "settled"
+    assert error.usage["prompt_tokens"] == 20 and error.usage["completion_tokens"] == 7
+    assert error.usage["cost"] is None and error.usage["cost_final"] is False
+    assert error.usage["claudexor"]["outcome"] == outcome
+    assert error.usage["claudexor"]["result_custody"]["state"] == "acknowledged"
+    classified = classify_llm_exception(error)
+    assert classified.kind == "provider_error" and not classified.retry_same_request
+    assert plan_next_wire_retry({}, error=error) is None
+    assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "settled"]
+    assert len(gateway.creates) == len(gateway.acks) == 1
+    assert not hasattr(error, "model_result")
+    public = json.dumps({"usage": error.usage, "ledger": ledger(root), "detail": gateway.detail(0)}) + caplog.text
+    public += "".join(path.read_text() for path in (root / "logs").glob("*.jsonl"))
+    for marker in (evidence["bodyBase64"], "private-error-marker", "private-stack-marker"):
+        assert marker not in public
+    assert retained(root)["failureEvidence"] == evidence
+
+
+def test_response_rejection_survives_existing_vision_ipc_reconstruction():
+    capture = ua.PhysicalAttemptCapture("attempt-one", MODEL, "claudexor", "settled", "opaque")
+    receipt = {"receipt_id": "receipt-one", "custody": None, "capture": asdict(capture),
+               "kind": "model", "text": "", "usage": {"prompt_tokens": 20}, "ledger_attempt_ids": ["attempt-one"],
+               "error": "", "problem": deepcopy(REJECTION), "operation_id": "operation-one", "model_role": "vision",
+               "route": deepcopy(ROUTE), "unknown": False, "control_reason": "", "model_result": None}
+    with pytest.raises(transport.ClaudexorModelError) as caught:
+        vision_process._decode_terminal(json.loads(json.dumps(receipt)), "receipt-one")
+    error = caught.value
+    assert error.code == "response_rejected" and error.stream_rejected and error.stream_incomplete
+    assert error.problem == REJECTION and error.operation_id == "operation-one" and error.route == ROUTE
+    assert error.physical_attempt_capture.state == "settled" and error.usage == receipt["usage"]
+    assert classify_llm_exception(error).kind == "provider_error"
+    assert not isinstance(error, ProviderNotDispatched)
+
+
+def test_unknown_diagnostic_display_keeps_only_compact_response_context():
+    error = transport.ClaudexorModelError({"code": "transport_unknown", "message": "Stream interrupted.",
+        "context": {"stage": "read", "errorCode": "UND_ERR_SOCKET", "requestId": "request-one",
+                    "vendorCode": "not-a-terminal-provider-fact", "stack": "private-stack-marker"}}, unknown=True)
+    assert error.display_message == (
+        "stage=read, cause=UND_ERR_SOCKET; model_outcome_unknown: Stream interrupted.")
+    assert "private-stack-marker" not in error.display_message
+    assert "request-one" not in error.display_message
+    assert error.code == "model_outcome_unknown" and not error.retryable
+
+
+def test_local_rejection_does_not_poison_the_next_account_preference(setup):
+    _, gateway, client = setup
+    gateway.results = [{**result(problem=deepcopy(REJECTION)), "message": None}, result()]
+    gateway.dispatch = ["response_received"] * 2
+    messages = [result()["message"]]
+    with pytest.raises(transport.ClaudexorModelError):
+        client.chat(messages, MODEL, cache_affinity="same-account-after-local-rejection")
+    client.chat(messages, MODEL, cache_affinity="same-account-after-local-rejection")
+    assert [payload["account"] for payload, _ in gateway.uploads] == [
+        {"mode": "auto", "preferredProfileId": "account-a"}] * 2
+
+
+def test_large_unknown_result_retains_exact_private_bytes_and_existing_pending_ack(setup):
+    root, gateway, client = setup
+    gateway.operation_catalog = [deepcopy(CAPTURE_OPERATION)]
+    body = b'\xff\xc3\x28' + b"x" * (4 * 1024 * 1024) + b"unparsed-suffix\r\n"
+    evidence = failure_evidence(body)
+    gateway.results = [{**result(outcome="unknown"), "message": None, "failureEvidence": evidence}]
+    gateway.dispatch = ["unknown"]
+    with pytest.raises(transport.ClaudexorModelError) as caught:
+        client.chat([], MODEL)
+    assert caught.value.code == "model_outcome_unknown"
+    assert not getattr(caught.value, "stream_rejected", False)
+    stored = retained(root)["failureEvidence"]
+    assert base64.b64decode(stored["bodyBase64"]) == body
+    assert stored["receivedBytes"] == len(body)
+    assert ledger(root)[-1]["state"] == "unresolved"
+    assert len(gateway.creates) == 1 and not gateway.acks

tests/test_claudexor_model_transport.py

@@ -427,3 +427,32 @@ def test_operation_identity_mismatch_is_not_an_accepted_response(gateway_factory
     with pytest.raises(cx.ClaudexorUnavailable) as raised:
         gateway.get_model_operation("op-one", timeout_sec=0.3)
     assert raised.value.code == "malformed_response"
+
+
+@pytest.mark.parametrize("capture", [False, True])
+def test_failure_evidence_opt_in_preserves_create_body(gateway_factory, capture):
+    wire = ModelWire()
+    gateway = gateway_factory(wire)
+    ref = gateway.upload_model_request(_request(), idempotency_key="evidence")
+    gateway.create_model_operation(ref, idempotency_key="evidence", capture_failure_evidence=capture)
+    request = wire.calls[-1]
+    assert request.url.query == (b"captureFailureEvidence=true" if capture else b"")
+    assert json.loads(request.content) == {"request": ref}
+    assert wire.payload == _bytes(_request())
+    assert wire.generation_count == 1
+
+
+def test_query_negotiation_uses_exact_wire_descriptor():
+    parameter = {"name": "captureFailureEvidence", "location": "query", "enum": ["true", "false"]}
+    operation = {"method": "POST", "path": "/v2/model-operations", "parameters": [parameter]}
+    assert cx.model_failure_evidence_supported([operation])
+    assert not cx.model_failure_evidence_supported([])
+    for change in ({"method": "GET"}, {"path": "/v2/model-operations/:id"}, {"parameters": []}):
+        assert not cx.model_failure_evidence_supported([{**operation, **change}])
+    for change in ({"name": "other"}, {"location": "header"}, {"enum": ["false"]}, {"enum": "true"}):
+        assert not cx.model_failure_evidence_supported([{**operation, "parameters": [{**parameter, **change}]}])
+    accounts = {"method": "GET", "path": "/v2/model-sources", "parameters": [
+        {"name": "view", "location": "query", "enum": ["accounts"]}]}
+    assert cx.account_catalog_supported([accounts], accounts["path"])
+    assert not cx.account_catalog_supported([operation], accounts["path"])
+    assert not cx.model_failure_evidence_supported([accounts])

tests/test_configured_model_recovery.py

@@ -0,0 +1,146 @@
+"""Supervising cognition recovers without replaying external execution."""
+
+import copy
+import hashlib
+import json
+
+import httpx
+import pytest
+
+from ouroboros import delegate_custody as custody, delegate_hold, loop, loop_transport
+from ouroboros import usage_accounting as ua
+from ouroboros.delegate_start_claims import claimed_start_request
+from tests.test_delegate_hold import _configured_registry, _loop_kwargs, _start_leaf
+from tests.test_transport_death_retry import _LedgerLLM, _ledger
+
+
+@pytest.mark.parametrize("external", ["inline", "consumed", "unread", "patch", "pending", "absent", "unreadable"])
+def test_saved_external_work_does_not_gate_supervising_cognition(tmp_path, monkeypatch, external):
+    """The incident's long consumed result and uncertain custody use one rail.
+
+    Only model I/O is scripted: physical accounting, custody replay, the main
+    loop and the fresh-start guard run as production code.
+    """
+    task_id, run_id = "t-death", "run-completed"
+    registry = _configured_registry(tmp_path, task_id)
+    output = "External result, including Unicode: готово.\n" * (1000 if external == "consumed" else 1)
+    if external in {"inline", "consumed", "unread", "patch"}:
+        custody._CUSTODY.pop(run_id, None)
+        row = custody.RunCustody(run_id=run_id, task_id=task_id, route_id="claude", model="external")
+        if external == "patch":
+            row.snapshot_id = "snapshot-preserved"
+        assert custody.record_started(tmp_path, row)
+        assert custody.settle_run(tmp_path, None, row, {"summary": {
+            "state": "succeeded", "spendUsd": 0, "spendEstimated": False,
+            "inputTokens": 5, "outputTokens": 5,
+        }})["settled"]
+        if external in {"consumed", "unread"}:
+            data = output.encode("utf-8")
+            artifact = tmp_path / "delegated_runs" / f"{run_id}.json"
+            artifact.parent.mkdir()
+            artifact.write_bytes(data)
+            row.output_artifact, row.output_sha, row.output_complete = (
+                f"delegated_runs/{run_id}.json", hashlib.sha256(data).hexdigest(), True)
+            assert custody.emit(tmp_path, custody.OUTPUT_SPILLED, {
+                "run_id": run_id, "task_id": task_id, "artifact": row.output_artifact,
+                "sha256": row.output_sha, "bytes": len(data), "staged": True, "full_content": True,
+            })
+            if external == "consumed":
+                assert custody.record_output_consumed(tmp_path, row, artifact=row.output_artifact,
+                    byte_length=len(data), sha256=row.output_sha, chars=len(output), lines=1001)
+        if external == "patch":
+            assert custody.record_patch_captured(tmp_path, row, patch_sha256="preserved-patch")
+    if external == "pending":
+        assert custody.record_start_requested(tmp_path, task_id=task_id,
+            invocation_id="pending-invocation", idempotency_key="pending-invocation", request={"prompt": "already sent"})
+    if external == "unreadable":
+        monkeypatch.setattr(custody, "custody_log_unreadable", lambda *_: True)
+
+    observations, messages_seen = [], []
+    class Model(_LedgerLLM):
+        def chat(self, **kwargs):
+            messages_seen.append(copy.deepcopy(kwargs["messages"]))
+            return super().chat(**kwargs)
+
+    model = Model(tmp_path, lambda: httpx.ReadError("host stream lost after external completion"))
+    monkeypatch.setattr(loop_transport, "upstream_transport_reachable",
+        lambda *a, **kw: observations.append(model.calls) or {"kind": "upstream_http", "status_code": 200})
+    monkeypatch.setattr(loop_transport, "interruptible_wait_sleep", lambda *a: False)
+    monkeypatch.setattr(delegate_hold, "_leaf_probe_live", lambda *a: pytest.fail("no holdable leaf"))
+    monkeypatch.setattr(custody, "release_task_runs", lambda *a: None)
+    monkeypatch.setenv("OUROBOROS_TASK_REVIEW_MODE", "off")
+    kwargs = _loop_kwargs(tmp_path, registry, [])
+    kwargs["llm"] = model
+    if external in {"inline", "consumed", "unread", "patch"}:
+        kwargs["messages"].extend([
+            {"role": "assistant", "content": None, "tool_calls": [{"id": "completed-wait", "type": "function",
+                "function": {"name": "delegate_wait", "arguments": json.dumps({"run_id": run_id})}}]},
+            {"role": "tool", "tool_call_id": "completed-wait", "content": output},
+        ])
+    custody_before = custody.event_log_path(tmp_path).read_bytes() if custody.event_log_path(tmp_path).exists() else b""
+    with ua.usage_scope(ua.UsageScope(drive_root=tmp_path, task_id=task_id, root_task_id=task_id, global_limit_usd=100)):
+        text, usage, trace = loop.run_llm_loop(**kwargs)
+
+    assert text == "done" and model.calls == 2 and observations == [1]
+    assert not trace["tool_calls"]  # No completed delegate/tool execution was replayed.
+    assert any("NEW physical model attempt" in str(row.get("content")) for row in messages_seen[-1])
+    if external in {"inline", "consumed", "unread", "patch"}:
+        assert [row["content"] for row in messages_seen[-1] if row.get("tool_call_id") == "completed-wait"] == [output]
+    ledger = _ledger(tmp_path)
+    assert [row["state"] for row in ledger] == ["reserved", "dispatched", "unresolved", "reserved", "dispatched", "settled"]
+    assert ledger[0]["attempt_id"] != ledger[3]["attempt_id"]
+    assert usage["transport_recovery"]["previous_attempt"]["physical_attempt_id"] == ledger[0]["attempt_id"]
+    assert ua.usage_projection(tmp_path)["unresolved_upper_bound_usd"] == 1.0
+    custody_after = custody.event_log_path(tmp_path).read_bytes() if custody.event_log_path(tmp_path).exists() else b""
+    rows = [json.loads(line) for line in custody_after.splitlines()]
+    assert [row for row in rows if row.get("type", "").startswith("delegate_")] == [
+        json.loads(line) for line in custody_before.splitlines() if json.loads(line).get("type", "").startswith("delegate_")]
+    if external in {"pending", "patch", "unreadable"}:
+        # Ability to think is not authority to duplicate the external work.
+        accepted, refusal = claimed_start_request(tmp_path, claim_target="", payload_busy=lambda *a: "",
+            actor_ctx=registry._ctx, enforce_actor_idle=True, task_id=task_id, invocation_id="duplicate")
+        assert not accepted
+        assert refusal["reason"] == ("replacement_custody_unknown" if external == "unreadable" else "replacement_requires_settlement")
+
+
+def test_live_hold_takes_over_an_existing_transport_episode(tmp_path, monkeypatch):
+    registry = _configured_registry(tmp_path)
+    calls, probes, snapshots = [], [], []
+    def send(_llm, messages, *args, **kwargs):
+        usage = args[8]
+        calls.append(len(calls) + 1)
+        snapshots.append(copy.deepcopy(messages))
+        if len(calls) <= 2:
+            if len(calls) == 2:
+                # This attempt now owns a live external leaf; the first did not.
+                _start_leaf(tmp_path)
+            usage.update(_last_llm_error_kind="provider_outcome_unknown",
+                         _pending_transport_outcome={"physical_attempt_id": f"old-{len(calls)}"})
+            return None, 0.0
+        usage.pop("_last_llm_error_kind", None)
+        return {"role": "assistant", "content": "integrated"}, 0.0
+
+    def observed(*args, **kwargs):
+        probes.append(len(calls))
+        assert len(calls) == 1, "A leaf wake must not require a new upstream observation"
+        return {"kind": "upstream_http", "status_code": 200}
+
+    monkeypatch.setattr(loop, "call_llm_with_retry", send)
+    monkeypatch.setattr(loop_transport, "upstream_transport_reachable", observed)
+    monkeypatch.setattr(loop_transport, "interruptible_wait_sleep", lambda *a: False)
+    monkeypatch.setattr(delegate_hold, "_leaf_probe_live", lambda *a: True)
+    monkeypatch.setattr(delegate_hold, "supervised_wait", lambda *a: json.dumps({
+        "status": "succeeded", "run_id": "run-leaf", "supervision_wake_id": "wake-after-hold"}))
+    monkeypatch.setattr(delegate_hold, "acknowledge_pending_wake", lambda *a, **kw: True)
+    monkeypatch.setattr(custody, "release_task_runs", lambda *a: None)
+    monkeypatch.setenv("OUROBOROS_TASK_REVIEW_MODE", "off")
+    result, usage, _trace = loop.run_llm_loop(**_loop_kwargs(tmp_path, registry, []))
+    assert result == "integrated" and len(calls) == 3 and probes == [1]
+    assert sum("NEW physical model attempt" in str(row.get("content")) for row in snapshots[-1]) == 1
+    assert any("[DELEGATED LEAF WAKE" in str(row.get("content")) for row in snapshots[-1])
+    events = [json.loads(line) for line in (tmp_path / "events.jsonl").read_text().splitlines()]
+    waits = [row for row in events if row.get("type") == "network_wait"]
+    assert [(row["phase"], row.get("detail", "")) for row in waits] == [
+        ("entered", ""), ("waiting", ""), ("recovered", "new_attempt_after_unknown_outcome"), ("ended", "hold_latched")]
+    assert waits[-1]["outcome_custody"]["physical_attempt_id"] == "old-1"
+    assert usage["transport_recovery"]["old_outcome"] == "unknown"

tests/test_deep_review_book_sources.py

@@ -11,7 +11,7 @@ from ouroboros.reference_books import compose_book, load_reference_book
 from tests.test_deep_review_slot import _native_row, _ScriptedLLM, _tool_call
 
 
-def _corpus(root):
+def _corpus(root, *, newline="\n"):
     files = {
         "BIBLE.md": "# Constitution\n\nThe constitutional source.\n",
         "docs/CHECKLISTS.md": "# Checklists\n\n## Review\n\nCheck the actual contract.\n",
@@ -22,10 +22,11 @@ def _corpus(root):
         for name in ("flow", "state"):
             files[f"docs/{book_id}/{name}.md"] = (
                 f"# {name.title()}\n\nIntroduction to {book_id} {name}.\n\n## Contract\n\nExact {book_id} {name} contract body.\n")
+    files = {rel: text.replace("\n", newline) for rel, text in files.items()}
     for rel, text in files.items():
         target = root / rel
         target.parent.mkdir(parents=True, exist_ok=True)
-        target.write_text(text)
+        target.write_bytes(text.encode("utf-8"))
     return files
 
 
@@ -37,13 +38,14 @@ def _required(root, rel):
             "range_basis": "unicode_text_universal_newlines"}
 
 
-def test_packed_chaptered_books_are_complete_once_and_stable_before_atlas(tmp_path, monkeypatch):
+@pytest.mark.parametrize("newline", ["\n", "\r\n"], ids=["lf", "crlf"])
+def test_packed_chaptered_books_are_complete_once_and_stable_before_atlas(tmp_path, monkeypatch, newline):
     monkeypatch.setattr(deep, "get_context_mode", lambda: "max")
     monkeypatch.setattr(deep, "_compute_graph_centrality", lambda *a: {})
     prefixes = []
     for name in ("first", "second"):
         repo, data = tmp_path / name, tmp_path / f"{name}-data"
-        files = _corpus(repo)
+        files = _corpus(repo, newline=newline)
         monkeypatch.setattr(deep, "_dulwich_tracked_paths", lambda *a: (list(files), []))
         monkeypatch.chdir(tmp_path)
         pack, stats = deep.build_review_pack(repo, data)

tests/test_deep_review_route_switch.py

@@ -97,7 +97,7 @@ def test_wide_route_switch_sends_once_and_reports_actual_model(packed):
     assert text.endswith(REPORT)
     assert any(model == MODEL_B for model, _, _ in packed.windows)
     assert packed.engine.uploads[-1][0]["messages"][-1]["content"] == PACK
-    assert len(packed.engine.operations) == 2
+    assert len(packed.engine.accepted_operations) == 2
     assert sum(row["state"] == "settled" for row in ledger(packed.root)) == 1
     assert packed.records[-1][0].model == MODEL_B
     assert packed.records[-1][0].session_profile == "account-b"
@@ -110,7 +110,7 @@ def test_subfloor_switch_refuses_before_new_send_and_keeps_prior_custody(packed,
     text, usage = packed.run()
     assert usage["execution_status"] == "infra_failed"
     assert "200,000" in text and "1,000,000" in text
-    assert len(packed.engine.operations) == 1
+    assert len(packed.engine.accepted_operations) == 1
     assert sum(row["state"] == "settled" for row in ledger(packed.root)) == int(prior_dispatch == "response_received")
     assert any(path.name.endswith("_model_response.json")
                for path in (packed.root / "observability" / "calls").rglob("*.json"))
@@ -125,7 +125,7 @@ def test_observed_only_subfloor_keeps_paid_report_with_actual_account(packed):
     assert "incomplete=none" not in text and text.endswith(REPORT)
     assert packed.windows[-1][2]["accountFingerprint"] == "identity-b"
     assert packed.records[-1][2]["status"] == "error"
-    assert len(packed.engine.operations) == 2
+    assert len(packed.engine.accepted_operations) == 2
     assert sum(row["state"] == "settled" for row in ledger(packed.root)) == 1
     custody = usage["claudexor"]["result_custody"]
     assert custody["state"] == "acknowledged"
@@ -147,7 +147,7 @@ def test_auto_actual_account_is_revalidated_without_another_generation(packed, w
     assert packed.engine.uploads[0][0]["account"] == {"mode": "auto"}
     assert packed.windows[-1][1] == "account-b"
     assert packed.windows[-1][2]["accountFingerprint"] == "identity-b"
-    assert len(packed.engine.operations) == 1
+    assert len(packed.engine.accepted_operations) == 1
     assert sum(row["state"] == "settled" for row in ledger(packed.root)) == 1
 
 
@@ -166,7 +166,7 @@ def test_changed_wide_route_still_checks_full_input_cap(packed, monkeypatch):
     text, usage = packed.run()
     assert usage["execution_status"] == "infra_failed"
     assert "input" in text.lower() and MODEL_B in text
-    assert len(packed.engine.operations) == 1
+    assert len(packed.engine.accepted_operations) == 1
 
 
 def test_resolve_packed_window_forwards_observed_account(monkeypatch):

tests/test_delegate_hold.py

@@ -3,8 +3,8 @@
 A configured-session nanny whose metered round dies ``provider_outcome_unknown``
 while EXACTLY one delegated leaf is alive must hold on the LEAF (zero provider
 calls) and resume with a wake-bearing NEW round — the unknown request is never
-resent. Control wakes and every ineligible shape keep today's no-resend
-terminal, and the terminal cleanup (leaf cancellation) fires only on terminals.
+resent. Control wakes keep their no-resend terminal. When no hold applies,
+ordinary managed recovery can restore the supervising model.
 """
 
 from __future__ import annotations
@@ -18,6 +18,7 @@ import pytest
 
 import ouroboros.delegate_hold as delegate_hold
 import ouroboros.loop as loop_mod
+import ouroboros.loop_transport as transport
 from ouroboros import delegate_custody as custody
 from ouroboros.delegate_supervision import read_unknown_hold, write_unknown_hold
 from ouroboros.loop import run_llm_loop
@@ -94,7 +95,21 @@ def _unknown_then_check_call(check):
     return fake_call, calls
 
 
+def _recover_model(monkeypatch):
+    monkeypatch.setattr(transport, "upstream_transport_reachable",
+                        lambda *a, **kw: {"kind": "upstream_http", "status_code": 200})
+    monkeypatch.setattr(transport, "interruptible_wait_sleep", lambda *a: False)
+
+    def recovered(_messages, usage):
+        usage.pop("_last_llm_error_kind", None)
+        return {"role": "assistant", "content": "recovered"}, 0.0
+
+    return recovered
+
+
 def test_unknown_with_live_leaf_holds_and_resumes_with_wake(tmp_path, monkeypatch, _quiet_probe):
+    monkeypatch.setattr(transport, "upstream_transport_reachable",
+                        lambda *a, **kw: pytest.fail("live hold must precede provider recovery"))
     wake_payload = {"status": "succeeded", "run_id": "run-leaf", "supervision_wake_id": "w1"}
     monkeypatch.setattr(delegate_hold, "supervised_wait",
                         lambda _ctx, _run: json.dumps(wake_payload))
@@ -127,6 +142,9 @@ def test_unknown_with_live_leaf_holds_and_resumes_with_wake(tmp_path, monkeypatc
     assert not read_unknown_hold(registry._ctx).get("run_id")  # inactive tombstone
     assert _quiet_probe == ["t-hold"]  # release only at the (successful) terminal
     assert any("holding on the leaf" in note for note in notes)
+    assert not any(json.loads(line).get("type") == "network_wait"
+                   for line in (tmp_path / "events.jsonl").read_text().splitlines())
+    assert "transport_recovery" not in usage
 
 
 def test_terminal_leaf_never_enters_hold(tmp_path, monkeypatch, _quiet_probe):
@@ -138,18 +156,17 @@ def test_terminal_leaf_never_enters_hold(tmp_path, monkeypatch, _quiet_probe):
     )
     monkeypatch.setattr(delegate_hold, "supervised_wait",
                         lambda *_a, **_k: pytest.fail("terminal leaf must not hold"))
-    fake_call, calls = _unknown_then_check_call(lambda *_: pytest.fail("no second dial"))
+    fake_call, calls = _unknown_then_check_call(_recover_model(monkeypatch))
     monkeypatch.setattr(loop_mod, "call_llm_with_retry", fake_call)
     monkeypatch.setenv("OUROBOROS_TASK_REVIEW_MODE", "off")
     monkeypatch.delenv("USE_LOCAL_FALLBACK", raising=False)
     registry = _configured_registry(tmp_path)
     _start_leaf(tmp_path)
     notes = []
-    _r, usage, trace = run_llm_loop(**_loop_kwargs(tmp_path, registry, notes))
+    result, usage, _trace = run_llm_loop(**_loop_kwargs(tmp_path, registry, notes))
 
-    assert calls["n"] == 1
-    assert usage.get("execution_status") == "infra_failed"
-    assert trace.get("forced_finalization", {}).get("source") == "provider_outcome_unknown_no_resend"
+    assert calls["n"] == 2 and result == "recovered"
+    assert usage["transport_recovery"]["old_outcome"] == "unknown"
     assert _read_hold_events(tmp_path) == []
 
 
@@ -219,31 +236,27 @@ def test_finalize_now_mid_hold_takes_no_call_terminal(tmp_path, monkeypatch, _qu
 def test_generic_task_and_multi_run_never_hold(tmp_path, monkeypatch, _quiet_probe):
     monkeypatch.setattr(delegate_hold, "supervised_wait",
                         lambda *_a, **_k: pytest.fail("ineligible shapes must not hold"))
-    fake_call, calls = _unknown_then_check_call(lambda *_: pytest.fail("no second dial"))
+    fake_call, calls = _unknown_then_check_call(_recover_model(monkeypatch))
     monkeypatch.setattr(loop_mod, "call_llm_with_retry", fake_call)
     monkeypatch.setenv("OUROBOROS_TASK_REVIEW_MODE", "off")
     monkeypatch.delenv("USE_LOCAL_FALLBACK", raising=False)
 
-    from ouroboros import loop_transport
-    monkeypatch.setattr(loop_transport, "interruptible_wait_sleep", lambda *args: False)
-    def exhausted_network_wait(episode, *, tools, **kwargs):
-        tools._ctx.task_metadata = {"deadline_at": "2000-01-01T00:00:00Z"}
-        return False
-    monkeypatch.setattr(loop_mod, "_continue_unknown_transport", exhausted_network_wait)
     # Generic tasks use the network owner, never the single-leaf nanny hold.
     registry = ToolRegistry(repo_dir=tmp_path, drive_root=tmp_path)
     registry._ctx.task_id = "t-generic"
     _start_leaf(tmp_path, task_id="t-generic", run_id="run-g")
-    _r, usage, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry, []))
-    assert calls["n"] == 1 and usage.get("execution_status") == "infra_failed"
+    result, usage, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry, []))
+    assert calls["n"] == 2 and result == "recovered"
+    assert usage["transport_recovery"]["old_outcome"] == "unknown"
 
     # Configured but TWO live leaves.
     calls["n"] = 0
     registry2 = _configured_registry(tmp_path, task_id="t-multi")
     _start_leaf(tmp_path, task_id="t-multi", run_id="run-m1")
     _start_leaf(tmp_path, task_id="t-multi", run_id="run-m2")
-    _r, usage2, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry2, []))
-    assert calls["n"] == 1 and usage2.get("execution_status") == "infra_failed"
+    result, usage2, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry2, []))
+    assert calls["n"] == 2 and result == "recovered"
+    assert usage2["transport_recovery"]["old_outcome"] == "unknown"
     assert _read_hold_events(tmp_path) == []
 
 
@@ -315,7 +328,7 @@ def test_repeated_unknown_reholds_with_backoff_floor(tmp_path, monkeypatch, _qui
 
 def test_refused_probe_and_state_less_payload_never_hold(tmp_path, monkeypatch, _quiet_probe):
     """A daemon refusal or a state-less payload is not evidence of a live leaf
-    (grok #1/#2, fable F3): the probe fails closed to today's terminal."""
+    and does not prohibit recovery of the supervising model."""
     import ouroboros.delegate_progress as progress_mod
 
     monkeypatch.setattr(delegate_hold, "supervised_wait",
@@ -327,19 +340,21 @@ def test_refused_probe_and_state_less_payload_never_hold(tmp_path, monkeypatch,
         raise RuntimeError("daemon unreachable")
 
     monkeypatch.setattr(progress_mod, "bounded_poll", raising_poll)
-    fake_call, calls = _unknown_then_check_call(lambda *_: pytest.fail("no second dial"))
+    fake_call, calls = _unknown_then_check_call(_recover_model(monkeypatch))
     monkeypatch.setattr(loop_mod, "call_llm_with_retry", fake_call)
     registry = _configured_registry(tmp_path, task_id="t-refused")
     _start_leaf(tmp_path, task_id="t-refused", run_id="run-r1")
-    _r, usage, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry, []))
-    assert calls["n"] == 1 and usage.get("execution_status") == "infra_failed"
+    result, usage, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry, []))
+    assert calls["n"] == 2 and result == "recovered"
+    assert usage["transport_recovery"]["old_outcome"] == "unknown"
 
     monkeypatch.setattr(progress_mod, "bounded_poll", lambda _gw, _run, _sec, **_k: {})
     calls["n"] = 0
     registry2 = _configured_registry(tmp_path, task_id="t-stateless")
     _start_leaf(tmp_path, task_id="t-stateless", run_id="run-r2")
-    _r, usage2, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry2, []))
-    assert calls["n"] == 1 and usage2.get("execution_status") == "infra_failed"
+    result, usage2, _t = run_llm_loop(**_loop_kwargs(tmp_path, registry2, []))
+    assert calls["n"] == 2 and result == "recovered"
+    assert usage2["transport_recovery"]["old_outcome"] == "unknown"
     assert _read_hold_events(tmp_path) == []
 
 

tests/test_file_resource_consistency.py

@@ -383,7 +383,7 @@ def test_wrapped_inline_read_uses_the_configured_supervisor(environment, monkeyp
     reg, ctx, _home, work, _data = environment
     ctx.task_constraint = TaskConstraint(mode='acting_subagent', surface='external_workspace', write_root=str(work))
     (work / '.env').write_text('FIXTURE_SECRET_MUST_NOT_REACH_OUTPUT', encoding='utf-8')
-    command = [sys.executable, '-c', "print(open('.env').read())"]
+    command = [sys.executable, '-c', "print(open('.env', encoding='utf-8').read())"]
     if wrapper in {'env', 'sh_env'}:
         command = ['env', *command]
     if wrapper in {'sh', 'sh_env'}:

tests/test_headless_workspace_shell.py

@@ -229,7 +229,7 @@ def test_workspace_shell_blocks_nested_symlink_escape_absolute_path(tmp_path, mo
     ctx.task_constraint = TaskConstraint(mode="acting_subagent", surface="external_workspace", write_root=str(workspace))
     registry.set_context(ctx)
 
-    result = registry.execute("run_command", {"cmd": f"touch {outlink / 'escaped.txt'}"})
+    result = registry.execute("run_command", {"cmd": ["touch", str(outlink / "escaped.txt")]})
     relative_result = registry.execute("run_command", {"cmd": "touch outlink/escaped-relative.txt"})
     bare_result = registry.execute("run_command", {"cmd": "touch outlink"})
     executable_name_result = registry.execute("run_command", {"cmd": ["touch", "touch"]})

tests/test_interpreter_family_write_fence.py

@@ -392,9 +392,9 @@ def test_readonly_python_string_method_executes_without_a_false_write(tmp_path,
 @pytest.mark.serial
 @pytest.mark.parametrize("body", [
     "from pathlib import Path\np = Path('.')\nprint((p / '.git').exists())\n"
-    "with (p / 'result.txt').open('a') as f: f.write('once\\n')\n",
+    "with (p / 'result.txt').open('a', newline='') as f: f.write('once\\n')\n",
     "from pathlib import Path\np = Path('nested')\np.mkdir(exist_ok=True)\n"
-    "with (p / 'result.txt').open('a') as f: f.write('once\\n')\n",
+    "with (p / 'result.txt').open('a', newline='') as f: f.write('once\\n')\n",
 ])
 def test_path_join_code_executes_once_without_guessing_a_control_write(tmp_path, monkeypatch, body):
     reg = _light_registry(tmp_path)

tests/test_llm_claudexor.py

@@ -127,7 +127,7 @@ class Gateway:
         self.results = results or [result()]
         self.dispatch = dispatch or ["response_received"] * len(self.results)
         self.uploads = []
-        self.operations = {}
+        self.accepted_operations = {}
         self.creates = []
         self.reads = []
         self.acks = []
@@ -138,17 +138,25 @@ class Gateway:
         self.pending = False
         self.read_error = False
         self.raw_result = None
+        self.operation_catalog = []
+        self.catalog_reads = 0
+        self.capture_requests = []
+
+    def operations(self):
+        self.catalog_reads += 1
+        return deepcopy(self.operation_catalog)
 
     def upload_model_request(self, payload, *, idempotency_key):
         self.uploads.append((deepcopy(payload), idempotency_key))
         return REF
 
-    def create_model_operation(self, ref, *, idempotency_key):
+    def create_model_operation(self, ref, *, idempotency_key, **options):
         assert ref == REF
+        self.capture_requests.append(deepcopy(options))
         self.creates.append(idempotency_key)
-        if idempotency_key not in self.operations:
-            self.operations[idempotency_key] = len(self.operations)
-        index = self.operations[idempotency_key]
+        if idempotency_key not in self.accepted_operations:
+            self.accepted_operations[idempotency_key] = len(self.accepted_operations)
+        index = self.accepted_operations[idempotency_key]
         if self.lose_create:
             self.lose_create = False
             raise ClaudexorUnavailable("daemon_unreachable", "lost create reply")
@@ -162,7 +170,8 @@ class Gateway:
 
     def detail(self, index):
         value = self.results[index]
-        return {"id": f"op-{index}", "state": "running" if self.pending else "succeeded" if value["outcome"] == "completed" else "failed",
+        succeeded = value["outcome"] == "completed" and value["message"] is not None
+        return {"id": f"op-{index}", "state": "running" if self.pending else "succeeded" if succeeded else "failed",
                 "dispatch": {"state": "started" if self.pending else self.dispatch[index], "route": value["route"]},
                 "response": {"state": "absent"} if self.pending else {"state": "ready", "ref": REF},
                 "problem": value["problem"]}
@@ -307,7 +316,7 @@ def test_lost_create_reply_rejoins_without_second_physical_attempt(setup):
     answer, usage = client.chat([{"role": "user", "content": "hi"}], MODEL)
     assert answer["content"] == "Ответ 🐍"
     assert len(gateway.creates) == 2 and gateway.creates[0] == gateway.creates[1]
-    assert len(gateway.operations) == 1 and len(usage["ledger_attempt_ids"]) == 1
+    assert len(gateway.accepted_operations) == 1 and len(usage["ledger_attempt_ids"]) == 1
     assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "settled"]
 
 
@@ -393,7 +402,7 @@ def test_control_connect_failure_after_acceptance_stays_unknown(setup):
     error = raised.value
     assert error.code == "model_outcome_unknown" and error.operation_id == "op-0"
     assert not is_pre_dispatch_transport_failure(error) and not is_retryable_transport_death(error)
-    assert ledger(root)[-1]["state"] == "unresolved" and len(gateway.operations) == 1
+    assert ledger(root)[-1]["state"] == "unresolved" and len(gateway.accepted_operations) == 1
     assert not gateway.acks
 
 
@@ -408,7 +417,7 @@ def test_proven_never_started_quota_attempts_do_not_spend_generation_limit(setup
         client.chat([], MODEL)
         with pytest.raises(ua.PhysicalAttemptLimitExceeded):
             client.chat([], MODEL)
-    assert len(gateway.operations) == 4
+    assert len(gateway.accepted_operations) == 4
     assert [r['state'] for r in ledger(root)].count('settled') == 1
 
 
@@ -420,7 +429,7 @@ def test_unknown_outcome_keeps_its_generation_limit_claim(setup):
             client.chat([], MODEL)
         with pytest.raises(ua.PhysicalAttemptLimitExceeded):
             client.chat([], MODEL)
-    assert len(gateway.operations) == 1
+    assert len(gateway.accepted_operations) == 1
 
 
 def test_confirmed_provider_failure_settles_real_usage_before_raising(setup):
@@ -466,7 +475,7 @@ def test_field_refusal_retains_then_acknowledges_once_with_display(setup, asynch
     assert f"provider_code={vendor}, parameter=input" in error.display_message[:220]
     assert error.physical_attempt_capture.state == "settled"
     assert error.usage["claudexor"]["result_custody"]["state"] == "acknowledged"
-    assert len(gateway.operations) == len(gateway.creates) == len(gateway.acks) == 1
+    assert len(gateway.accepted_operations) == len(gateway.creates) == len(gateway.acks) == 1
     assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "settled"]
 
 
@@ -484,7 +493,7 @@ def test_proven_not_started_releases_and_never_fabricates_provider_usage(setup,
     assert raised.value.physical_attempt_capture.provider_error_type == (vendor or "ClaudexorModelNotDispatched")
     assert gateway.uploads[0][0]["options"]["temperature"] == 0.2
     assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "released"]
-    assert len(gateway.operations) == 1
+    assert len(gateway.accepted_operations) == 1
 
 
 def test_typed_subject_refusal_suppresses_next_auto_preference(setup):
@@ -519,7 +528,7 @@ def test_native_reset_requires_actual_account_change_and_keeps_canonical_tools(s
     if not set(change) & {"credentialProfileId", "accountFingerprint"}:
         with pytest.raises(transport.ClaudexorModelNotDispatched):
             client.chat(messages, MODEL)
-        assert len(gateway.operations) == 1
+        assert len(gateway.accepted_operations) == 1
     else:
         _, usage = client.chat(messages, MODEL)
         assert len(usage["ledger_attempt_ids"]) == 2
@@ -540,7 +549,7 @@ def test_ack_failure_preserves_paid_result_and_does_not_repeat(setup, error):
     answer, usage = client.chat([{"role": "user", "content": "hi"}], MODEL)
     assert answer == result()["message"] and retained(root) == result()
     assert usage["claudexor"]["result_custody"]["state"] == "pending"
-    assert len(gateway.operations) == 1 and ledger(root)[-1]["state"] == "settled"
+    assert len(gateway.accepted_operations) == 1 and ledger(root)[-1]["state"] == "settled"
 
 
 def test_failed_local_result_retention_withholds_ack_but_keeps_answer(setup, monkeypatch):
@@ -586,7 +595,7 @@ def test_async_tools_and_capture_remain_in_callers_context(setup):
         assert answer == result()["message"]
 
     asyncio.run(run())
-    assert len(gateway.operations) == 1 and ledger(root)[-1]["state"] == "settled"
+    assert len(gateway.accepted_operations) == 1 and ledger(root)[-1]["state"] == "settled"
 
 
 def test_gigachat_async_tools_still_refuse_before_provider_io(setup, monkeypatch):
@@ -685,7 +694,7 @@ def test_observer_failure_does_not_lose_response_or_repeat_generation(setup):
 
     answer, _ = client.chat([], MODEL, model_operation_observer=failed)
     assert answer == result()["message"] and retained(root) == result()
-    assert len(gateway.operations) == 1 and ledger(root)[-1]["state"] == "settled"
+    assert len(gateway.accepted_operations) == 1 and ledger(root)[-1]["state"] == "settled"
 
 
 def test_async_local_switch_projects_its_new_capture_to_the_caller(setup, monkeypatch):
@@ -758,7 +767,7 @@ def test_unknown_engine_outcome_retains_response_without_resend_or_false_zero(se
         client.chat([{"role": "user", "content": "hi"}], MODEL)
     assert raised.value.code == "model_outcome_unknown" and not raised.value.type
     assert retained(root) == gateway.results[0] and not gateway.acks
-    assert len(gateway.operations) == 1 and ledger(root)[-1]["state"] == "unresolved"
+    assert len(gateway.accepted_operations) == 1 and ledger(root)[-1]["state"] == "unresolved"
     assert ledger(root)[-1].get("cost_usd") is None
 
 
@@ -770,7 +779,7 @@ def test_continuation_repair_is_bounded_to_one_unstarted_operation(setup):
     gateway.dispatch = ["not_started", "not_started"]
     with pytest.raises(transport.ClaudexorModelNotDispatched):
         client.chat([result()["message"]], MODEL)
-    assert len(gateway.operations) == 2
+    assert len(gateway.accepted_operations) == 2
     assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "released"] * 2
 
 
@@ -783,7 +792,7 @@ def test_unreleased_attempt_cannot_authorize_continuation_repair(setup, monkeypa
     with pytest.raises(transport.ClaudexorModelNotDispatched) as raised:
         client.chat([result()["message"]], MODEL)
     assert raised.value.physical_attempt_capture.state == "unresolved"
-    assert len(gateway.operations) == 1 and ledger(root)[-1]["state"] == "unresolved"
+    assert len(gateway.accepted_operations) == 1 and ledger(root)[-1]["state"] == "unresolved"
 
 
 def test_model_switch_strips_native_envelope_but_not_tool_results():
@@ -802,7 +811,7 @@ def test_caller_control_interrupts_pending_operation_without_false_success(setup
     gateway.pending = True
     with pytest.raises(transport.ClaudexorModelError) as raised:
         client.chat([{"role": "user", "content": "hi"}], MODEL,
-                    model_poll_control=lambda: "deadline_exceeded" if gateway.operations else None)
+                    model_poll_control=lambda: "deadline_exceeded" if gateway.accepted_operations else None)
     assert raised.value.control_reason == "deadline_exceeded"
     assert gateway.cancels == [("op-0", "host_cancelled")]
     assert ledger(root)[-1]["state"] == "unresolved" and not gateway.acks
@@ -873,7 +882,7 @@ def test_missing_old_account_does_not_authorize_native_reset(setup):
     gateway.dispatch = ["not_started"]
     with pytest.raises(transport.ClaudexorModelNotDispatched):
         client.chat([message], MODEL)
-    assert len(gateway.operations) == 1
+    assert len(gateway.accepted_operations) == 1
 
 
 def test_caller_control_before_create_proves_no_dispatch(setup):

tests/test_llm_provider_golden.py

@@ -460,7 +460,7 @@ def _install_model_operation_fake(stack: contextlib.ExitStack, recorder: _Record
 
     class ModelGateway(Gateway):
         def create_model_operation(self, ref, *, idempotency_key):
-            if idempotency_key not in self.operations:
+            if idempotency_key not in self.accepted_operations:
                 payload = self.uploads[-1][0]
                 step = recorder.record("claudexor.model_operation", payload=payload)
                 if step.get("kind") == "error" and step.get("code") != "provider_policy_refusal":
@@ -469,7 +469,7 @@ def _install_model_operation_fake(stack: contextlib.ExitStack, recorder: _Record
                     "code": step.get("code"), "message": step.get("message")}, route={
                     "source": payload["source"], "model": payload["model"],
                     "credentialProfileId": "conformance-profile", "accountFingerprint": "conformance-identity"}))
-                index = len(self.operations)
+                index = len(self.accepted_operations)
                 if index == 0:
                     self.results[0], self.dispatch[0] = value, "response_received"
                 else:
@@ -612,7 +612,7 @@ def _observe(spec: Dict[str, Any]) -> Dict[str, Any]:
         client = LLMClient(**client_args)
         call = copy.deepcopy(spec["call"])
         if spec.get("cancel_model_after_create"):
-            call["kwargs"]["model_poll_control"] = lambda: "cancelled" if recorder.model_gateway.operations else None
+            call["kwargs"]["model_poll_control"] = lambda: "cancelled" if recorder.model_gateway.accepted_operations else None
         try:
             result = _call_route(client, call)
         except BaseException as exc:  # noqa: BLE001 - the raise IS the projection
@@ -637,7 +637,7 @@ def _observe(spec: Dict[str, Any]) -> Dict[str, Any]:
     observed["unused_script_steps"] = len(recorder.script)
     if spec.get("model_operation"):
         gateway = recorder.model_gateway
-        observed["model_control"] = {"create_posts": len(gateway.creates), "operations": len(gateway.operations),
+        observed["model_control"] = {"create_posts": len(gateway.creates), "operations": len(gateway.accepted_operations),
                                      "unique_create_keys": len(set(gateway.creates)), "cancels": gateway.cancels}
     return _jsonable(observed)
 

tests/test_model_wait.py

@@ -325,7 +325,7 @@ def test_async_cancellation_resolves_only_its_wait_and_keeps_shared_task(live_wa
         assert terminal["resolution"] == "caller_cancelled"
 
     asyncio.run(run())
-    assert not controller.closed and len(transport.operations) == 1
+    assert not controller.closed and len(transport.accepted_operations) == 1
     assert all(row["state"] == "resolved" for row in load_task_result(root, "task-one")["model_waits"].values())
 
 
@@ -623,7 +623,7 @@ def test_unproved_pool_cause_never_enters_resource_wait(live_wait, pool_context)
     transport.dispatch = ["not_started"]
     with pytest.raises(ClaudexorModelError, match="credential_pool_exhausted"):
         client.chat([{"role": "user", "content": "Do not infer quota"}], MODEL, model_role="main")
-    assert not controller.waits and events.empty() and len(transport.operations) == 1
+    assert not controller.waits and events.empty() and len(transport.accepted_operations) == 1
 
 
 def test_mixed_pool_cannot_bypass_unknown_physical_custody(live_wait):
@@ -636,7 +636,7 @@ def test_mixed_pool_cannot_bypass_unknown_physical_custody(live_wait):
     transport.dispatch = ["unknown"]
     with pytest.raises(ClaudexorModelError, match="model_outcome_unknown"):
         client.chat([{"role": "user", "content": "No duplicate generation"}], MODEL, model_role="main")
-    assert not controller.waits and events.empty() and len(transport.operations) == 1
+    assert not controller.waits and events.empty() and len(transport.accepted_operations) == 1
 
 
 def test_mixed_pool_wait_keeps_calendar_deadline_and_existing_quota_union(live_wait, monkeypatch):
@@ -691,14 +691,14 @@ def test_call_can_decline_resource_wait_without_losing_task_binding(live_wait, c
     assert error.physical_attempt_capture.state == "released"
     assert error.model_role_route == {"role": "light", "model": MODEL, "use_local": False,
                                      "credential_profile_id": ""}
-    assert not controller.waits and events.empty() and len(transport.operations) == 1
+    assert not controller.waits and events.empty() and len(transport.accepted_operations) == 1
     assert transport.uploads[0][0]["account"] == {"mode": "auto"}
     assert "wait_for_resources" not in json.dumps(transport.uploads[0][0])
     assert model_wait.current_model_wait() is controller and not controller.closed
     assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "released"]
 
     answer, usage = call()
-    assert answer == result()["message"] and len(transport.operations) == 3
+    assert answer == result()["message"] and len(transport.accepted_operations) == 3
     assert len(usage["ledger_attempt_ids"]) == 2
     assert any(row["state"] == "waiting" for row in list(events.queue))
     assert controller.overrides["light"]["model_account_override"] == ""
@@ -716,4 +716,4 @@ def test_declining_resource_wait_still_honors_owner_control(live_wait, monkeypat
         else:
             client.chat(*args, **kwargs)
     assert caught.value.control_reason == "finalize_requested"
-    assert not transport.operations and not controller.waits and events.empty()
+    assert not transport.accepted_operations and not controller.waits and events.empty()

tests/test_model_wait_controls.py

@@ -121,14 +121,14 @@ def test_pinned_wait_rejects_other_catalog_account_before_new_generation(live_wa
     polls = []
 
     def catalog(source, profile=None, **kwargs):
-        assert profile == "account-a" and len(gateway.operations) == 1
+        assert profile == "account-a" and len(gateway.accepted_operations) == 1
         polls.append(profile)
         return {"source": source, "credentialProfileId": "account-b" if len(polls) == 1 else "account-a",
                 "models": [{"id": "exact-model"}]}
 
     monkeypatch.setattr(client, "claudexor_model_catalog", catalog)
     client.chat([], MODEL, model_role="light")
-    assert len(polls) == 2 and len(gateway.operations) == 2
+    assert len(polls) == 2 and len(gateway.accepted_operations) == 2
     assert all(payload["account"] == {"mode": "pin", "profileId": "account-a"} for payload, _key in gateway.uploads)
 
 
@@ -170,7 +170,7 @@ def test_main_wait_does_not_call_configured_api_fallback_before_owner_switch(mai
                                            before_dispatch=_candidate_before_dispatch(request_body, request))
 
     def catalog(*args, **kwargs):
-        assert api_calls == [] and len(gateway.operations) == 1
+        assert api_calls == [] and len(gateway.accepted_operations) == 1
         row = next(event for event in reversed(list(events.queue)) if event.get("type") == "task_model_wait")
         response = decide({"request_id": "switch-api", "decision_id": f"model_wait:task-one:{row['wait_id']}",
                            "revision": row["revision"], "action": "switch", "model": "openai::alternate",
@@ -184,7 +184,7 @@ def test_main_wait_does_not_call_configured_api_fallback_before_owner_switch(mai
         ctx.messages, tools, ctx.llm, ctx.drive_logs, lambda *_args, **_kwargs: None, queue.Queue(),
         task_id="task-one", drive_root=ctx.drive_root, event_queue=events)
     assert text == "Finished" and api_calls == ["openai"]
-    assert usage["_model_route"] == {} and len(gateway.operations) == 1
+    assert usage["_model_route"] == {} and len(gateway.accepted_operations) == 1
     assert any(message.get("content") == "verified read A" for message in ctx.messages)
     assert any(message.get("content") == "completed review B" for message in ctx.messages)
 
@@ -437,7 +437,7 @@ def test_real_main_control_preserves_candidate_without_new_summary(main_call, mo
         task_id="task-one", drive_root=ctx.drive_root, event_queue=events)
     assert len(held) == 1 and text == completed["message"]["content"]
     assert usage["reason_code"] == trace["forced_finalization"]["reason_code"] == expected_reason
-    assert len(gateway.operations) == 2  # Paid answer + interrupted call, never a summary retry.
+    assert len(gateway.accepted_operations) == 2  # Paid answer + interrupted call, never a summary retry.
     assert trace["forced_finalization"]["source"].startswith("model_wait_retained_candidate")
     if stop == "wrap_unknown":
         assert usage["_last_llm_error_kind"] == "provider_outcome_unknown"
@@ -510,7 +510,7 @@ def test_hard_cancel_returns_empty_events_to_real_worker_loop_and_keeps_queue_ow
             return None  # End the test's worker only after verifying retained ownership.
 
     worker_process.worker_main(1, Input(), events, str(ctx.drive_root), str(ctx.drive_root))
-    assert len(reads) == 2 and len(gateway.operations) == 1 and not crashes
+    assert len(reads) == 2 and len(gateway.accepted_operations) == 1 and not crashes
     assert ledger(ctx.drive_root)[-1]["state"] == "unresolved"
 
 

tests/test_nano_ui_static.py

@@ -4,8 +4,8 @@ ROOT = Path(__file__).parents[1]
 
 
 def test_settings_and_chat_expose_nano_context_mode():
-    settings = (ROOT / 'web/modules/settings_ui.js').read_text()
-    chat = (ROOT / 'web/modules/chat.js').read_text()
+    settings = (ROOT / 'web/modules/settings_ui.js').read_text(encoding="utf-8")
+    chat = (ROOT / 'web/modules/chat.js').read_text(encoding="utf-8")
     assert "{ value: 'nano', label: 'Nano' }" in settings
     assert 'data-mode="nano">Nano' in chat
     assert "['nano', 'low', 'max'].includes(data.context_mode)" in chat
@@ -13,8 +13,8 @@ def test_settings_and_chat_expose_nano_context_mode():
 
 
 def test_chat_control_ids_match_instance_wiring_and_nano_has_active_style():
-    chat = (ROOT / 'web/modules/chat.js').read_text()
-    styles = (ROOT / 'web/style.css').read_text()
+    chat = (ROOT / 'web/modules/chat.js').read_text(encoding="utf-8")
+    styles = (ROOT / 'web/style.css').read_text(encoding="utf-8")
     # The markup keeps the chat-* ids that byId() resolves for the main
     # instance (and namespaces for project panels), so handlers cannot drift
     # to a selector for a different control.

tests/test_processing_claudexor.py

@@ -66,7 +66,7 @@ def test_response_received_with_proved_no_generation_reprepares_standard(setup,
     _message, usage = asyncio.run(client.chat_async(**kwargs)) if asynchronous else client.chat(**kwargs)
     first, second = [row[0] for row in gateway.uploads]
     assert {**first, "options": {**first["options"], "processingPreference": "standard"}} == second
-    assert len(calls) == 1 and len(gateway.operations) == 2 and len(gateway.acks) == 2
+    assert len(calls) == 1 and len(gateway.accepted_operations) == 2 and len(gateway.acks) == 2
     finals = list({row["attempt_id"]: row for row in ledger(root)}.values())
     assert [row["state"] for row in finals] == ["released", "settled"]
     assert finals[0]["candidate_raw_sha256"] != finals[1]["candidate_raw_sha256"]
@@ -94,7 +94,7 @@ def test_only_explicit_no_start_advisory_proof_allows_retry(setup, axis):
             cx.chat_claudexor(target, [], None, service_tier="flex")
         else:
             client.chat([], MODEL, **kwargs)
-    assert len(gateway.operations) == 1
+    assert len(gateway.accepted_operations) == 1
     if axis.startswith("unknown"):
         assert raised.value.code == "model_outcome_unknown" and ledger(root)[-1]["state"] == "unresolved"
     if axis == "exact_native":
@@ -143,7 +143,7 @@ def test_one_repair_on_each_axis_uses_the_same_bounded_preparation_loop(setup, o
     gateway.results = [value for value, _dispatch in pair] + [{**result(), "processing": receipt()}]
     gateway.dispatch = [dispatch for _value, dispatch in pair] + ["response_received"]
     client.chat([result()["message"]], MODEL, processing_preference="economy")
-    assert len(gateway.operations) == 3
+    assert len(gateway.accepted_operations) == 3
     assert [row["state"] for row in {r["attempt_id"]: r for r in ledger(root)}.values()] == ["released", "released", "settled"]
 
 
@@ -167,7 +167,7 @@ def test_standard_must_be_supported_before_emitting_a_processing_retry(setup, mo
     gateway.results = [refusal()]
     with pytest.raises(cx.ClaudexorModelNotDispatched):
         client.chat([], MODEL, processing_preference="economy")
-    assert len(gateway.operations) == 1, "Do not retry by omitting unsupported Standard and inheriting native premium"
+    assert len(gateway.accepted_operations) == 1, "Do not retry by omitting unsupported Standard and inheriting native premium"
 
 
 def test_old_facade_does_not_send_new_query_parameter(monkeypatch):

tests/test_release_proof.py

@@ -311,7 +311,7 @@ def test_every_future_release_receipt_requires_real_embedded_betterleaks():
 def test_future_final_artifact_lanes_smoke_betterleaks_from_the_artifact():
     workflow = (REPO / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")
     build_job = workflow[
-        workflow.index("  build:") : workflow.index("  vendor-package-smoke:")
+        workflow.index("\n  build:\n") : workflow.index("\n  vendor-package-smoke:\n")
     ]
     assert build_job.count("scripts/betterleaks_platform_smoke.py") == 4
     assert '--bundle-root "$MOUNT/Ouroboros.app/Contents/Resources"' in build_job
@@ -411,7 +411,7 @@ def test_linux_package_smoke_starts_the_desktop_launcher_on_ubuntu_22_04():
 def test_vendor_distro_smoke_is_informational_and_never_gates_a_release():
     workflow = (REPO / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")
     vendor_job = workflow[
-        workflow.index("  vendor-package-smoke:") : workflow.index("  release:")
+        workflow.index("\n  vendor-package-smoke:\n") : workflow.index("\n  release:\n")
     ]
     assert "continue-on-error: true" in vendor_job
     assert "smoke_linux_packages.sh vendor" in vendor_job
@@ -419,14 +419,14 @@ def test_vendor_distro_smoke_is_informational_and_never_gates_a_release():
     # The gating lane runs every package through Docker Hub images only, so a
     # vendor registry outage cannot stop a tagged release.
     build_job = workflow[
-        workflow.index("  build:") : workflow.index("  vendor-package-smoke:")
+        workflow.index("\n  build:\n") : workflow.index("\n  vendor-package-smoke:\n")
     ]
     assert "smoke_linux_packages.sh official" in build_job
     assert "smoke_linux_packages.sh vendor" not in build_job
 
     release_needs = next(
         line
-        for line in workflow[workflow.index("  release:") :].splitlines()
+        for line in workflow[workflow.index("\n  release:\n") :].splitlines()
         if line.strip().startswith("needs:")
     )
     assert "vendor-package-smoke" not in release_needs
@@ -535,7 +535,7 @@ def test_release_workflow_orders_smoke_sbom_attestation_and_draft_verification()
     assert "$env:HOMEDRIVE = Split-Path -Qualifier $HomeDir" in workflow
     assert "$env:HOMEPATH = $HomeDir.Substring" in workflow
     build_job = workflow[
-        workflow.index("  build:") : workflow.index("  vendor-package-smoke:")
+        workflow.index("\n  build:\n") : workflow.index("\n  vendor-package-smoke:\n")
     ]
     job_env = build_job[build_job.index("    env:") : build_job.index("    steps:")]
     assert "BUILD_CERTIFICATE_BASE64:" not in job_env

tests/test_review_model_override.py

@@ -93,11 +93,11 @@ def test_native_wait_switch_rechecks_bound_before_send_and_never_replays_read(li
         with pytest.raises(ReviewRouteUnavailable) as raised:
             executor.execute()
         assert raised.value.code == "native_transcript_cap_exceeded"
-        assert len(gateway.operations) == 2
+        assert len(gateway.accepted_operations) == 2
         assert executor.failure_custody()["native_transcript_bound"] == 0
         with pytest.raises(ReviewRouteUnavailable):
             executor.execute()
-        assert len(gateway.operations) == 2
+        assert len(gateway.accepted_operations) == 2
     else:
         answer = executor.execute()
         assert answer.raw_text == final["message"]["content"]
@@ -106,7 +106,7 @@ def test_native_wait_switch_rechecks_bound_before_send_and_never_replays_read(li
         sent = gateway.uploads[-1][0]
         assert sent["account"] == {"mode": "pin", "profileId": "account-b"}
         assert any(message.get("role") == "tool" and "completed original read" in message["content"] for message in sent["messages"])
-        assert answer.usage["native_rounds"] == 2 and len(gateway.operations) == 3
+        assert answer.usage["native_rounds"] == 2 and len(gateway.accepted_operations) == 3
     assert len(reads) == len(decisions) == 1
     assert controller.overrides.keys() == {"reviewer:critic"}
     assert [row["state"] for row in ledger(root)].count("settled") == (1 if narrow else 2)

tests/test_review_temperature_defaults.py

@@ -63,7 +63,7 @@ def test_raw_default_hint_defers_but_explicit_temperature_stays_strict(setup, mo
         assert raised.value.code == "unsupported_parameter"
         assert gateway.uploads[0][0]["options"]["temperature"] == explicit
         assert ledger(root)[-1]["state"] == "released"
-    assert len(gateway.operations) == 1
+    assert len(gateway.accepted_operations) == 1
     assert "default_temperature" not in gateway.uploads[0][0]["options"]
     assert kwargs["temperature"] is explicit and kwargs["default_temperature"] == 0.2
 
@@ -132,7 +132,7 @@ def test_wait_switch_restores_api_hint_and_later_override_defers_again(live_wait
     # A reused caller originally naming API must not carry its resolved 0.2 as explicit.
     kwargs["model"] = "anthropic::other-model"
     _message, usage = call()
-    assert usage["prompt_tokens"] == 20 and len(gateway.operations) == 2
+    assert usage["prompt_tokens"] == 20 and len(gateway.accepted_operations) == 2
     assert all("temperature" not in payload["options"] for payload, _key in gateway.uploads)
     assert [row["state"] for row in ledger(root)].count("released") == 1
 
@@ -179,7 +179,7 @@ def test_actual_review_authors_reach_strict_raw_dispatch(setup, monkeypatch, sur
     assert "temperature" not in payload["options"]
     assert "default_temperature" not in payload["options"]
     assert payload["account"] == {"mode": "pin", "profileId": "account-a"}
-    assert len(gateway.operations) == 1 and ledger(root)[-1]["state"] == "settled"
+    assert len(gateway.accepted_operations) == 1 and ledger(root)[-1]["state"] == "settled"
 
 
 def test_review_custody_distinguishes_hint_from_explicit_without_changing_old_keys():
@@ -212,4 +212,4 @@ def test_explicit_review_temperature_beats_both_host_hints(setup, monkeypatch, n
         client.chat(**kwargs)
     assert raised.value.code == "unsupported_parameter"
     assert gateway.uploads[0][0]["options"]["temperature"] == expected
-    assert len(gateway.operations) == 1 and ledger(root)[-1]["state"] == "released"
+    assert len(gateway.accepted_operations) == 1 and ledger(root)[-1]["state"] == "released"

tests/test_run_script_capability.py

@@ -14,12 +14,12 @@ from ouroboros.tools.registry import ToolContext, ToolRegistry
 
 pytestmark = pytest.mark.serial
 
-_PYTHON = "import sys\nfrom pathlib import Path\nprint(*sys.argv[1:], sep='\\n')\nwith Path('once.txt').open('a') as f: f.write('once\\n')\n"
+_PYTHON = "import sys\nfrom pathlib import Path\nprint(*sys.argv[1:], sep='\\n')\nwith Path('once.txt').open('a', newline='') as f: f.write('once\\n')\n"
 _SCRIPTS = {
     "python3": _PYTHON,
     "python_absolute": _PYTHON,
     "node": "console.log(process.argv.slice(2).join('\\n')); require('node:fs').appendFileSync('once.txt', 'once\\n');",
-    "perl": "print join(\"\\n\", @ARGV), \"\\n\"; open(my $f, '>>', 'once.txt') or die $!; print {$f} \"once\\n\"; close($f);",
+    "perl": "print join(\"\\n\", @ARGV), \"\\n\"; open(my $f, '>>', 'once.txt') or die $!; binmode($f); print {$f} \"once\\n\"; close($f);",
     "zsh": "printf '%s\\n' \"$@\"\nprintf '%s\\n' once >> once.txt\n",
     "lua": "print(table.concat(arg, '\\n')); local f = assert(io.open('once.txt', 'a')); f:write('once\\n'); f:close()",
 }

tests/test_runtime_mode_elevation.py

@@ -1793,10 +1793,12 @@ def test_run_shell_writes_skill_state_examples(shape, filename, tmp_path, monkey
         # ``bash`` on hosted Windows resolves to WSL, which has no distro in
         # the GitHub runner image. Exercise the same redirect/copy shapes via
         # the native command interpreter instead of depending on WSL setup.
+        # Separate operands let subprocess quote paths without backslash-escaping
+        # the inner quotes of a prequoted cmd body.
         cmd = (
-            ["cmd.exe", "/d", "/c", f'type "{payload}" > "{target}"']
+            ["cmd.exe", "/d", "/c", "type", str(payload), ">", str(target)]
             if shape == "redirect"
-            else ["cmd.exe", "/d", "/c", f'copy /Y "{payload}" "{target}" >NUL']
+            else ["cmd.exe", "/d", "/c", "copy", "/Y", str(payload), str(target), ">NUL"]
         )
     else:
         cmd = (

tests/test_services_tool_v2.py

@@ -790,7 +790,7 @@ def test_light_service_python_body_executes_once_with_bound_environment(tmp_path
     monkeypatch.setattr("ouroboros.safety.check_safety", supervisor)
     body = ("import os, time\nfrom pathlib import Path\n"
             f"p = Path({selected_path!r})\np.parent.mkdir(parents=True, exist_ok=True)\n"
-            "with p.open('a') as output: output.write('once\\n')\n"
+            "with p.open('a', newline='') as output: output.write('once\\n')\n"
             + "\n".join(f"assert os.environ[{key!r}] == {child_env[key]!r}" for key in
                         ("HOME", "USERPROFILE", "OUROBOROS_DATA_DIR", "OUROBOROS_SETTINGS_PATH"))
             + "\nprint('READY', flush=True)\ntime.sleep(60)\n")

tests/test_skill_exec.py

@@ -371,7 +371,7 @@ def test_run_shell_blocks_actual_self_authored_marker_write_pre_exec(tmp_path, m
 
     result = registry.execute(
         "run_command",
-        {"cmd": ["bash", "-c", f"printf '{{}}' > {marker}"]},
+        {"cmd": ["bash", "-c", f"printf '{{}}' > {shlex.quote(str(marker))}"]},
     )
 
     assert "WORKSPACE_SHELL_BLOCKED" in result and "process was not started" in result

tests/test_skill_review_persist_guard.py

@@ -340,6 +340,16 @@ def test_direct_review_ignores_historical_completed_job_hash(tmp_path, monkeypat
     assert load_review_state(drive_root, "alpha").content_hash == content_hash
 
 
+async def _wait_for_reconcile(task, started):
+    # Preparation includes filesystem work and imports, not a two-second contract.
+    # Still surface early lifecycle completion instead of waiting for a phase it skipped.
+    while not started.is_set():
+        if task.done():
+            result = await task
+            raise AssertionError(f"lifecycle finished before extension reconcile: {result!r}")
+        await asyncio.sleep(0.01)
+
+
 def test_cancellation_during_extension_reconcile_keeps_lifecycle_lane(tmp_path, monkeypatch):
     from ouroboros.skill_review import SkillReviewOutcome
     from ouroboros.skill_review_runner import run_skill_review_lifecycle
@@ -375,7 +385,7 @@ def test_cancellation_during_extension_reconcile_keeps_lifecycle_lane(tmp_path,
 
     def fake_reconcile(*_args, **_kwargs):
         reconcile_started.set()
-        release_reconcile.wait(2)
+        release_reconcile.wait()
         return reconcile_receipt("extension_loaded", "ok")
 
     monkeypatch.setattr(runner, "_reconcile_extension_payload", fake_reconcile)
@@ -384,30 +394,33 @@ def test_cancellation_during_extension_reconcile_keeps_lifecycle_lane(tmp_path,
         task = asyncio.create_task(
             run_skill_review_lifecycle(ctx, "alpha", source="test", review_impl=fake_review)
         )
-        assert await asyncio.to_thread(reconcile_started.wait, 2)
-        task.cancel()
-        await asyncio.sleep(0.05)
-        task.cancel()
-        await asyncio.sleep(0.05)
-        active = lifecycle_queue.queue_snapshot()["active"]
-        assert active is not None
-        assert active["target"] == "alpha"
-        quick = asyncio.create_task(
-            lifecycle_queue.run_lifecycle_job(
-                kind="review",
-                target="beta",
-                dedupe_key="review:beta:hash",
-                runner=lambda: asyncio.sleep(0, result={"quick": True}),
-                options=lifecycle_queue.LifecycleJobOptions(drive_root=drive_root),
+        try:
+            await _wait_for_reconcile(task, reconcile_started)
+            task.cancel()
+            await asyncio.sleep(0.05)
+            task.cancel()
+            await asyncio.sleep(0.05)
+            active = lifecycle_queue.queue_snapshot()["active"]
+            assert active is not None
+            assert active["target"] == "alpha"
+            quick = asyncio.create_task(
+                lifecycle_queue.run_lifecycle_job(
+                    kind="review",
+                    target="beta",
+                    dedupe_key="review:beta:hash",
+                    runner=lambda: asyncio.sleep(0, result={"quick": True}),
+                    options=lifecycle_queue.LifecycleJobOptions(drive_root=drive_root),
+                )
             )
-        )
-        await asyncio.sleep(0.05)
-        assert not quick.done()
-        release_reconcile.set()
-        result = await asyncio.wait_for(task, timeout=2)
-        assert result["status"] == "clean"
-        assert await asyncio.wait_for(quick, timeout=2) == {"quick": True}
-        assert lifecycle_queue.queue_snapshot()["active"] is None
+            await asyncio.sleep(0.05)
+            assert not quick.done()
+            release_reconcile.set()
+            result = await asyncio.wait_for(task, timeout=2)
+            assert result["status"] == "clean"
+            assert await asyncio.wait_for(quick, timeout=2) == {"quick": True}
+            assert lifecycle_queue.queue_snapshot()["active"] is None
+        finally:
+            release_reconcile.set()
 
     asyncio.run(main())
 
@@ -451,7 +464,7 @@ def test_heartbeat_continues_during_extension_reconcile(tmp_path, monkeypatch):
 
     def fake_reconcile(*_args, **_kwargs):
         reconcile_started.set()
-        release_reconcile.wait(2)
+        release_reconcile.wait()
         return reconcile_receipt("extension_loaded", "ok")
 
     monkeypatch.setattr(runner, "_reconcile_extension_payload", fake_reconcile)
@@ -460,22 +473,25 @@ def test_heartbeat_continues_during_extension_reconcile(tmp_path, monkeypatch):
         task = asyncio.create_task(
             run_skill_review_lifecycle(ctx, "alpha", source="test", review_impl=fake_review)
         )
-        assert await asyncio.to_thread(reconcile_started.wait, 2)
-        job_path = review_job_state_path(drive_root, "alpha")
-        before = json.loads(job_path.read_text(encoding="utf-8"))["last_heartbeat_at"]
-        # A bounded wait, not 20 x 10 ms: on windows-latest the heartbeat's atomic replace can
-        # lose a few rounds to this very poll holding the file open (sharing violation, logged
-        # and retried by the beat), and the clock ticks at ~15.6 ms — 200 ms saw no change.
-        for _ in range(60):
-            await asyncio.sleep(0.05)
-            after = json.loads(job_path.read_text(encoding="utf-8"))["last_heartbeat_at"]
-            if after != before:
-                break
-        assert after != before, "no heartbeat within 3 s while the reconcile blocks"
-        release_reconcile.set()
-        result = await asyncio.wait_for(task, timeout=2)
-        assert result["status"] == "clean"
-        final = json.loads(job_path.read_text(encoding="utf-8"))
-        assert final["status"] == "completed"
+        try:
+            await _wait_for_reconcile(task, reconcile_started)
+            job_path = review_job_state_path(drive_root, "alpha")
+            before = json.loads(job_path.read_text(encoding="utf-8"))["last_heartbeat_at"]
+            # A bounded wait, not 20 x 10 ms: on windows-latest the heartbeat's atomic replace can
+            # lose a few rounds to this very poll holding the file open (sharing violation, logged
+            # and retried by the beat), and the clock ticks at ~15.6 ms — 200 ms saw no change.
+            for _ in range(60):
+                await asyncio.sleep(0.05)
+                after = json.loads(job_path.read_text(encoding="utf-8"))["last_heartbeat_at"]
+                if after != before:
+                    break
+            assert after != before, "no heartbeat within 3 s while the reconcile blocks"
+            release_reconcile.set()
+            result = await asyncio.wait_for(task, timeout=2)
+            assert result["status"] == "clean"
+            final = json.loads(job_path.read_text(encoding="utf-8"))
+            assert final["status"] == "completed"
+        finally:
+            release_reconcile.set()
 
     asyncio.run(main())

tests/test_snapshot_file_inputs.py

@@ -1,4 +1,4 @@
-"""Copied binary/large inputs retain exact preimages without entering Git's ODB."""
+"""Snapshot working bytes stay exact; binary/large preimages stay outside Git's ODB."""
 from hashlib import sha256
 from pathlib import Path
 import subprocess
@@ -172,3 +172,112 @@ def test_baseline_file_copy_failure_cleans_checkout_ref_and_record(tmp_path, mon
     assert find_execution_snapshot("input-snapshot", tmp_path / "data") is None
     assert not git(target, "for-each-ref", "refs/ouroboros/delegated/").stdout
     assert not list((tmp_path / "snapshots").glob("dlg_*"))
+
+
+@pytest.mark.parametrize("newline", [b"\n", b"\r\n"], ids=["lf", "crlf"])
+@pytest.mark.parametrize("policy", ["autocrlf", "attributes"])
+def test_git_checkout_preserves_working_bytes_and_normal_patch_semantics(tmp_path, newline, policy):
+    target = target_tree(tmp_path)
+    git(target, "config", "core.autocrlf", "true" if policy == "autocrlf" else "false")
+    if policy == "attributes":
+        (target / ".gitattributes").write_bytes(b"*.txt text eol=crlf\n")
+    original = newline.join([b"original", b"working bytes", b""])
+    for name in ("code.txt", "staged.txt", "untracked.txt"):
+        (target / name).write_bytes(original)
+    git(target, "add", "staged.txt")
+    (target / "deleted.txt").write_bytes(b"removed before snapshot\n")
+    git(target, "add", "deleted.txt")
+    (target / "deleted.txt").unlink()
+    before_status = git(target, "status", "--porcelain=v1", "-z").stdout
+    before_index = git(target, "ls-files", "--stage", "-z").stdout
+    before_head = git(target, "rev-parse", "HEAD").stdout
+
+    handle = snapshot(tmp_path, target)
+    execution = Path(handle.path)
+    for name in ("code.txt", "staged.txt", "untracked.txt"):
+        assert (execution / name).read_bytes() == original
+    assert not (execution / "deleted.txt").exists()
+    assert git(target, "status", "--porcelain=v1", "-z").stdout == before_status
+    assert git(target, "ls-files", "--stage", "-z").stdout == before_index
+    assert git(target, "rev-parse", "HEAD").stdout == before_head
+    assert all((target / name).read_bytes() == original
+               for name in ("code.txt", "staged.txt", "untracked.txt"))
+    manifest, rows = capture(tmp_path, handle)
+    assert manifest["status"] == "ready_no_changes" and manifest["patch_size"] == 0 and rows == []
+
+    changed = original.replace(b"original", b"child edit")
+    (execution / "code.txt").write_bytes(changed)
+    manifest, rows = capture(tmp_path, handle)
+    assert manifest["status"] == "ready_with_changes" and rows == []
+    assert manifest["tracked_changed"] == ["code.txt"]
+    patch = tmp_path / "capture" / "workspace.patch"
+    git(target, "apply", "--binary", "--check", str(patch))
+    git(target, "apply", "--binary", str(patch))
+    # Explicit apply retains the target's normal Git checkout representation.
+    assert (target / "code.txt").read_bytes() == b"child edit\r\nworking bytes\r\n"
+    assert (target / "staged.txt").read_bytes() == original
+    assert (target / "untracked.txt").read_bytes() == original
+    assert git(target, "ls-files", "--stage", "-z").stdout == before_index
+    assert git(target, "rev-parse", "HEAD").stdout == before_head
+
+
+@pytest.mark.parametrize("failure", ["copy_error", "source_changed"])
+def test_regular_input_copy_failure_cleans_snapshot_without_rewriting_target(tmp_path, monkeypatch, failure):
+    from ouroboros import artifacts
+
+    target = target_tree(tmp_path)
+    before_index = git(target, "ls-files", "--stage", "-z").stdout
+    before_head = git(target, "rev-parse", "HEAD").stdout
+    before_bytes = (target / "code.txt").read_bytes()
+    original = artifacts.copy_artifact_file
+    newer = b"concurrent owner edit\n"
+
+    def copy(source, destination, **kwargs):
+        if Path(source) == target / "code.txt":
+            if failure == "copy_error":
+                raise OSError("source changed during copy")
+            Path(source).write_bytes(newer)
+        return original(source, destination, **kwargs)
+
+    monkeypatch.setattr(artifacts, "copy_artifact_file", copy)
+    expected = OSError if failure == "copy_error" else subprocess.CalledProcessError
+    with pytest.raises(expected):
+        snapshot(tmp_path, target)
+    assert (target / "code.txt").read_bytes() == (newer if failure == "source_changed" else before_bytes)
+    assert git(target, "ls-files", "--stage", "-z").stdout == before_index
+    assert git(target, "rev-parse", "HEAD").stdout == before_head
+    assert find_execution_snapshot("input-snapshot", tmp_path / "data") is None
+    assert not git(target, "for-each-ref", "refs/ouroboros/delegated/").stdout
+    assert not list((tmp_path / "snapshots").glob("dlg_*"))
+
+
+def test_snapshot_keeps_git_link_entries_out_of_regular_file_copy(tmp_path, monkeypatch):
+    from ouroboros import artifacts
+
+    target = target_tree(tmp_path)
+    # Git's portable symlink checkout is a regular file containing the target;
+    # the baseline mode, not its host file type, still owns that representation.
+    git(target, "config", "core.symlinks", "false")
+    (target / "link").write_bytes(b"code.txt")
+    oid = git(target, "hash-object", "-w", "link").stdout.decode().strip()
+    git(target, "update-index", "--add", "--cacheinfo", f"120000,{oid},link")
+    commit = git(target, "rev-parse", "HEAD").stdout.decode().strip()
+    git(target, "update-index", "--add", "--cacheinfo", f"160000,{commit},nested")
+    git(target, "-c", "user.name=Fixture", "-c", "user.email=f@invalid", "commit", "-m", "link entries")
+    (target / "nested").mkdir()  # An uninitialized gitlink stays Git-owned.
+    before_index = git(target, "ls-files", "--stage", "-z").stdout
+    original = artifacts.copy_artifact_file
+
+    def copy(source, destination, **kwargs):
+        assert Path(source).name not in {"link", "nested"}
+        return original(source, destination, **kwargs)
+
+    monkeypatch.setattr(artifacts, "copy_artifact_file", copy)
+    handle = snapshot(tmp_path, target)
+    execution = Path(handle.path)
+    assert (execution / "link").read_bytes() == b"code.txt"
+    assert git(execution, "ls-tree", "HEAD", "link").stdout.startswith(b"120000 blob ")
+    assert git(execution, "ls-tree", "HEAD", "nested").stdout.startswith(b"160000 commit ")
+    assert git(target, "ls-files", "--stage", "-z").stdout == before_index
+    manifest, rows = capture(tmp_path, handle)
+    assert manifest["status"] == "ready_no_changes" and rows == []

tests/test_subscription_main_wait.py

@@ -91,7 +91,7 @@ def test_native_account_repair_rebinds_real_physical_candidate_before_send(main_
     assert dispatched[1]["physical_context"]["route_fp"] == "capacity-account-b"
     assert dispatched[1]["physical_context"]["capacity_total_tokens"] == 240_000
     assert observations[0]["model_route"] == ROUTE_B
-    assert len(gateway.operations) == 2 and gateway.creates[0] != gateway.creates[1]
+    assert len(gateway.accepted_operations) == 2 and gateway.creates[0] != gateway.creates[1]
     resent = gateway.uploads[1][0]["messages"]
     assert "nativeContinuation" not in resent[2]
     assert resent[2]["tool_calls"] == original[2]["tool_calls"] and resent[3:] == original[3:]
@@ -106,7 +106,7 @@ def test_quota_auto_wait_rejoins_same_round_then_repairs_changed_account(main_ca
     gateway.dispatch = ["not_started", "not_started", "response_received"]
     answer, cost, mode = _dispatch(ctx)
     assert answer and cost is None and mode == "max"
-    assert len(gateway.operations) == 3
+    assert len(gateway.accepted_operations) == 3
     assert ctx.accumulated_usage["rounds"] == 1
     assert ctx.accumulated_usage["_model_route"] == ROUTE_B
     assert ctx.accumulated_usage["_context_route_fp"] == "capacity-account-b"
@@ -178,14 +178,14 @@ def test_manual_switch_updates_only_waiting_role_and_continues_current_main_call
 def test_main_control_interrupt_is_typed_no_retry_and_keeps_operation_custody(main_call, monkeypatch):
     ctx, gateway, controller, _events, _decide, _observations = main_call
     gateway.pending = True
-    monkeypatch.setattr(controller, "control_reason", lambda: "cancelled" if gateway.operations else None)
+    monkeypatch.setattr(controller, "control_reason", lambda: "cancelled" if gateway.accepted_operations else None)
     with pytest.raises(model_wait.ModelWaitInterrupted) as raised:
         _dispatch(ctx)
     error = raised.value
     assert error.control_reason == "cancelled" and error.operation_id == "op-0"
     assert error.physical_attempt_capture.state == "unresolved"
     assert error.model_role_route["role"] == "main"
-    assert len(gateway.operations) == 1 and gateway.cancels == [("op-0", "host_cancelled")]
+    assert len(gateway.accepted_operations) == 1 and gateway.cancels == [("op-0", "host_cancelled")]
     assert not controller.waits and not ctx.accumulated_usage.get("_last_llm_retry_same_request")
 
 
@@ -198,7 +198,7 @@ def test_cancel_after_result_keeps_settled_usage_and_exact_result(main_call, mon
     assert raised.value.usage["prompt_tokens"] == 20
     assert raised.value.physical_attempt_capture.state == "settled"
     assert raised.value.route == ROUTE
-    assert len(gateway.operations) == 1
+    assert len(gateway.accepted_operations) == 1
 
 
 def test_native_repair_without_main_callback_refuses_stale_physical_fit(setup):
@@ -212,7 +212,7 @@ def test_native_repair_without_main_callback_refuses_stale_physical_fit(setup):
     with ua.bind_physical_attempt_context(physical):
         with pytest.raises(model_wait.ModelWaitInterrupted, match="model_wait_reprepare_required"):
             client.chat([result()["message"]], MODEL, model_role="main")
-    assert len(gateway.operations) == 1
+    assert len(gateway.accepted_operations) == 1
 
 
 def test_unknown_main_outcome_never_retries_or_waits_for_quota(main_call):
@@ -220,7 +220,7 @@ def test_unknown_main_outcome_never_retries_or_waits_for_quota(main_call):
     gateway.results, gateway.dispatch = [result(outcome="unknown")], ["unknown"]
     answer, _cost, _mode = _dispatch(ctx)
     assert answer is None and ctx.accumulated_usage["_last_llm_error_kind"] == "provider_outcome_unknown"
-    assert len(gateway.operations) == 1 and not controller.waits
+    assert len(gateway.accepted_operations) == 1 and not controller.waits
     assert ledger(ctx.drive_root)[-1]["state"] == "unresolved"
 
 
@@ -312,7 +312,7 @@ def test_a_same_route_wait_and_reprepare_update_the_original_slot(main_call, tur
         assert _dispatch(ctx)[0]
     # The slot the loop still owns is the one the durable result must have replaced.
     assert ctx.tools._ctx.model_turn_state is slot and slot.envelope == TURN
-    assert len(gateway.operations) == 2 and gateway.uploads[-1][0]["nativeContinuation"] is None
+    assert len(gateway.accepted_operations) == 2 and gateway.uploads[-1][0]["nativeContinuation"] is None
 
 
 def test_a_helper_call_cannot_overwrite_the_running_loop_slot(setup, turn_engine):
@@ -535,7 +535,7 @@ def test_live_owner_wait_reprojects_affinity(main_call, monkeypatch, destination
         "execution_id": ctx.accumulated_usage["execution_id"],
         "owner_switch_saved": decisions[0]["saved"],
         "completed_tool_texts": [x["content"] for x in ctx.messages if x.get("role") == "tool"],
-        "gateway_operations": len(gateway.operations),
+        "gateway_operations": len(gateway.accepted_operations),
     }
     assert facts["completed_tool_texts"] == ["verified read A", "completed review B"]
     assert ctx.accumulated_usage["execution_id"] == CACHE_REPREPARE_EXECUTION
@@ -568,7 +568,7 @@ def test_recorded_wait_override_reprojects_affinity_before_send(main_call, initi
     # API/local have no subscription quota wait of their own.
     controller.overrides["main"] = {"model": MODEL, "use_local": False, "model_account_override": ""}
     answer, _, _ = _dispatch(ctx)
-    assert answer and len(gateway.operations) == 1
+    assert answer and len(gateway.accepted_operations) == 1
     payload = gateway.uploads[0][0]
     facts = {
         "initial_model": initial_model,
@@ -576,7 +576,7 @@ def test_recorded_wait_override_reprojects_affinity_before_send(main_call, initi
         "active_model": ctx.active_model,
         "cache_key": payload["options"].get("cacheKey"),
         "execution_id": ctx.accumulated_usage["execution_id"],
-        "gateway_operations": len(gateway.operations),
+        "gateway_operations": len(gateway.accepted_operations),
         "completed_tool_texts": [x["content"] for x in ctx.messages if x.get("role") == "tool"],
     }
     assert facts["completed_tool_texts"] == ["verified read A", "completed review B"]

tests/test_subscription_vision_capability.py

@@ -74,7 +74,7 @@ def test_preset_images_reach_real_main_transport(subscription_transport, catalog
     assert gateway.uploads[0][0]["account"] == {"mode": "pin", "profileId": "explicit-main"}
     assert calls == [("codex", "explicit-main", "exact-model")]
     assert messages == original
-    assert len(gateway.creates) == 1 and len(gateway.operations) == 1
+    assert len(gateway.creates) == 1 and len(gateway.accepted_operations) == 1
     assert [row["state"] for row in ledger(root)] == ["reserved", "dispatched", "settled"]
 
 

tests/test_transport_repeat_owner_stop.py

@@ -1,5 +1,6 @@
 """A cooperative stop interrupts only the unsent paid transport-repeat grant."""
 
+import json
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
@@ -168,9 +169,9 @@ def test_paid_repeat_empty_peek_reuses_existing_wait_proof(tmp_path, monkeypatch
     assert ctx._loop_mailbox_seen_ids == {"old"}
 
 
-@pytest.mark.parametrize("with_leaf", [False, True])
-def test_wrapup_reason_survives_the_live_delegate_hold(tmp_path, monkeypatch, with_leaf):
-    from ouroboros import claudexor_daemon, delegate_custody, delegate_progress
+@pytest.mark.parametrize("with_leaf, during_hold", [(False, False), (True, False), (True, True)])
+def test_wrapup_reason_survives_the_live_delegate_hold(tmp_path, monkeypatch, with_leaf, during_hold):
+    from ouroboros import claudexor_daemon, delegate_custody, delegate_progress, delegate_hold
     from tests.test_delegate_hold import _configured_registry, _start_leaf, _loop_kwargs as hold_kwargs
 
     monkeypatch.setenv("OUROBOROS_TASK_REVIEW_MODE", "off")
@@ -181,12 +182,23 @@ def test_wrapup_reason_survives_the_live_delegate_hold(tmp_path, monkeypatch, wi
     if with_leaf:
         _start_leaf(tmp_path, task_id="t-death", run_id="fixture-leaf")
 
-    def death():
+    def request_wrapup():
         intent = cancel_intents.request_cancel(tmp_path, "t-death", requested_stop_policy=cancel_intents.STOP_POLICY_FINALIZE)
         owner_mailbox.write_owner_message(tmp_path, REASON_OWNER_REQUESTED_FINALIZATION, "t-death",
             msg_id=owner_stop_control_id(intent), kind=owner_mailbox.KIND_FINALIZE_NOW)
+
+    def death():
+        if not during_hold:
+            request_wrapup()
         return httpx.ReadError("controlled post-dispatch failure")
 
+    def hold(*args):
+        request_wrapup()
+        return json.dumps({"status": "progress", "wake_events": [{"kind": "finalize_now"}]})
+
+    if during_hold:
+        monkeypatch.setattr(delegate_hold, "supervised_wait", hold)
+
     llm = _LedgerLLM(tmp_path, death)
     kwargs = hold_kwargs(tmp_path, registry, [])
     kwargs["llm"] = llm

tests/test_transport_unknown_continuation.py

@@ -226,14 +226,15 @@ def test_subscription_requires_fresh_typed_upstream_observation(monkeypatch, fac
     assert bool(transport.upstream_transport_reachable(None, "claudexor::codex=test", timeout=3)) is (fact == "upstream")
 
 
-def test_claudexor_control_loss_keeps_same_operation_past_read_window(tmp_path, monkeypatch):
+@pytest.mark.parametrize("route_kind", ["", "agent_session"])
+def test_claudexor_control_loss_keeps_same_operation_past_read_window(tmp_path, monkeypatch, route_kind):
     from ouroboros import llm_claudexor
     from ouroboros.gateways.claudexor import ClaudexorUnavailable
     inv = llm_claudexor._ModelInvocation({"usage_model": "claudexor::codex=test"}, {}, {"timeout": 1})
     inv.operation_id, inv.invocation_id, inv.task_id, inv.root = "same-op", "same-attempt", "t", tmp_path
     inv.create_attempted = True
     now, reads = [0.0], []
-    ctx = SimpleNamespace(task_id="t")
+    ctx = SimpleNamespace(task_id="t", _configured_subagent_route_kind=route_kind)
     waiter = SimpleNamespace(tool_context=ctx, control_reason=lambda: None)
     monkeypatch.setattr(llm_claudexor, "current_model_wait", lambda: waiter)
     monkeypatch.setattr(llm_claudexor, "time", SimpleNamespace(monotonic=lambda: now[0], sleep=lambda t: now.__setitem__(0, now[0]+t)))
@@ -389,7 +390,10 @@ def test_upstream_head_uses_connection_window_in_every_socket_phase(monkeypatch,
     assert requests[0].extensions["timeout"] == dict(connect=bound, read=bound, write=bound, pool=bound)
 
 
-def test_unknown_policy_keeps_configured_session_nanny_out_of_managed_continuation(tmp_path):
+def test_unknown_policy_admits_configured_session_model_to_managed_continuation(tmp_path):
     ctx = SimpleNamespace(task_id="t", exact_model_route=True, _configured_subagent_route_kind="agent_session")
-    from ouroboros import loop_transport
-    assert loop_transport.reconcile_transport_wait(None, ctx, msg_present=False, error_kind="provider_outcome_unknown", drive_logs=tmp_path, task_id="t", model="m", emit_progress=lambda *a, **kw: pytest.fail("unexpected automatic continuation")) is None
+    episode = transport.reconcile_transport_wait(None, ctx, msg_present=False,
+        error_kind="provider_outcome_unknown", drive_logs=tmp_path, task_id="t", model="m",
+        emit_progress=lambda *a, **kw: None)
+    assert episode is not None and episode.wait_cause == "provider_outcome_unknown"
+    assert not episode.interactive

tests/test_vision_model_wait.py

@@ -34,6 +34,9 @@ def _install_child_fixture(mode, events_path):
         polls = 0
         lost = False
 
+        def operations(self):
+            return []  # Legacy model-operation catalog.
+
         def upload_model_request(self, payload, *, idempotency_key):
             event("upload", key=idempotency_key, payload=payload)
             return REF

tests/test_workspace_write_shape.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import os
 import pathlib
+import shlex
 
 import pytest
 
@@ -498,7 +500,7 @@ def test_glued_operator_is_not_a_path_candidate(tmp_path):
 
     assert check(["sh", "-c", "git reset HEAD scratch/ 2>/dev/null; rm -rf scratch/"]) is None
     assert check(["sh", "-c", "node build.js 2>/dev/null; echo ok"]) is None
-    redirect_block = check(["sh", "-c", f"node t.js > {outside / 'out.log'} 2>&1"]) or ""
+    redirect_block = check(["sh", "-c", f"node t.js > {shlex.quote(str(outside / 'out.log'))} 2>&1"]) or ""
     assert "WORKSPACE_SHELL_BLOCKED" in redirect_block
     assert str(outside / "out.log") in redirect_block
 
@@ -559,7 +561,7 @@ def test_round6_redirect_file_targets_outside_workspace_are_blocked(tmp_path):
     outside = tmp_path / "outside"
     outside.mkdir()
     (workspace / "1").symlink_to(outside / "numeric-log")
-    for body in (f"echo x >& {outside / 'redirect-log'}", "echo x >1"):
+    for body in (f"echo x >& {shlex.quote(str(outside / 'redirect-log'))}", "echo x >1"):
         args = {"cmd": ["sh", "-c", body], "cwd": str(workspace)}
         out = _shell_guard_text(reg, args, "advanced") or ""
         assert "WORKSPACE_SHELL_BLOCKED" in out, (body, out)
@@ -592,14 +594,14 @@ def test_round3_old_block_coverage_stays_blocked(tmp_path):
     outside.mkdir()
 
     commands = (
-        ("cd_relative_write", f"cd {outside} && echo x > rel"),
-        ("pushd_relative_write", f"pushd {outside} && echo x > rel"),
-        ("awk_redirect", f"awk '{{print $1}}' input > {outside / 'awk'}"),
-        ("rsync_destination", ["rsync", "src", str(outside / "rsync")]),
+        ("cd_relative_write", f"cd {shlex.quote(str(outside))} && echo x > rel"),
+        ("pushd_relative_write", f"pushd {shlex.quote(str(outside))} && echo x > rel"),
+        ("awk_redirect", f"awk '{{print $1}}' input > {shlex.quote(str(outside / 'awk'))}"),
+        ("rsync_destination", ["rsync", "src", pathlib.Path(os.path.relpath(outside / "rsync", workspace)).as_posix()]),
         ("tar_chdir_extract", ["tar", "-C", str(outside), "-xf", "a.tar"]),
-        ("append_redirect", f"echo x >> {outside / 'append'}"),
-        ("stderr_redirect", f"awk '{{print $1}}' input 2> {outside / 'stderr'}"),
-        ("combined_redirect", f"awk '{{print $1}}' input &> {outside / 'combined'}"),
+        ("append_redirect", f"echo x >> {shlex.quote(str(outside / 'append'))}"),
+        ("stderr_redirect", f"awk '{{print $1}}' input 2> {shlex.quote(str(outside / 'stderr'))}"),
+        ("combined_redirect", f"awk '{{print $1}}' input &> {shlex.quote(str(outside / 'combined'))}"),
     )
     if pathlib.Path(r"C:\outside\drive.txt").is_absolute():
         # These are physical absolute targets on Windows; on POSIX the
@@ -761,13 +763,13 @@ def test_outside_root_block_names_the_spelling_the_model_used(tmp_path):
     alias.symlink_to(real, target_is_directory=True)
     for spelled in (f"{alias / 'f'}", "../outside_real/f"):
         text = _shell_guard_text(reg,
-            {"cmd": ["sh", "-c", f"echo x > {spelled}"], "cwd": str(workspace)}, "advanced",
+            {"cmd": ["sh", "-c", f"echo x > {shlex.quote(spelled)}"], "cwd": str(workspace)}, "advanced",
         ) or ""
         assert "WORKSPACE_SHELL_BLOCKED" in text and str(workspace) in text, spelled
         assert spelled in text or str(real / "f") in text
     # Identical spellings are named once.
     text = _shell_guard_text(reg,
-        {"cmd": ["sh", "-c", f"echo x > {real / 'g'}"], "cwd": str(workspace)}, "advanced",
+        {"cmd": ["sh", "-c", f"echo x > {shlex.quote(str(real / 'g'))}"], "cwd": str(workspace)}, "advanced",
     ) or ""
     assert "WORKSPACE_SHELL_BLOCKED" in text and str(real / "g") in text