Merge pull request #1870 from razzant/contrib/operation-truth-1825-20261014

Preserve invocation truth and completion after nested failures

docs/architecture/06-agent-core.md

@@ -649,9 +649,9 @@ Warm and lane-retaining cold refuse a selected queued root needing that lane; un
 
 `POST /api/tasks/{id}/pause` takes a live root and text-free `{request_id}`; Stop/cancel veto. Fence persistence precedes ACK/latch; Q probes never wait: pre-write failure refuses, post-write contention yields `latch_pending`. New actions use new ids; Resume retains history. Local critics keep a warm park; others use exact pause. One `stop_task_owned` request stops external runs, never treating requested/unknown as death. Accepted cold holders use that off-loop owner without a sleep `external_writers_running` flag. Resume keeps source, money, deadline and custody controls.
 
-`run_tool_handler` takes the prepared callable under the root launch lock with current Stop/Pause and inherited deadlines, then invokes it on the same worker outside that lock. Ordinary queueing starts preparation only, with no second executor. Sticky calls keep their one-use ticket/thread affinity; nested sends, including skill child spawn, take fresh gates. `owner_pause._control_projection` shares root/member owner-pause, sleep and status reads: capture the signature before strict load, cache only under that original still-current signature, never a later stamp on old bytes. Cached checks need no body decode between writes. Unknown authority refuses; contention retries proven-unsent claims off Q/root/money locks with the same reservation/consumer, never the handler.
+`run_tool_handler` admits the prepared callable under root Stop/Pause/deadlines, then runs on the same worker outside admission; queueing is preparation, not a second executor. Sticky tickets keep thread affinity; nested sends, including skill child spawn, take fresh gates. A phase mutex orders close/start without DB waits; refusals preserve invocation entry and physical cause. `owner_pause._control_projection` caches strict reads under the original still-current signature; unchanged checks decode no body. Unknown refuses; unsent contention retries off Q/root/money locks with the same reservation/consumer, never the handler.
 
-`budget_pause` owns the actual Future under frozen root/task/attempt/invocation identity, caller pin and real callback holds. Host-error settlement precedes completion; passive callbacks are no hold. Joined Future-owned calls need no eager SQL claim. `tool_custody.py` retains the same id before caller escape or opaque/process/control effects; local return settles no independent process/run/merge/receipt/money duty. Unbound callers keep durable fallback; legacy rows remain readable. Failed publication joins body/real holds outside locks/store budgets before caller release. Direct Main Stop is cooperative, never thread kill; pooled termination and Panic remain independent.
+`budget_pause` owns the Future's frozen root/task/attempt/invocation identity, caller pin/real holds; passive callbacks are no hold. Host errors settle before completion; joins need no eager SQL claim. `tool_custody.py` retains it before escape/effects; return settles no independent duty. Exact idempotent settlement retries only contention in the store budget, without launch locks/body replay; persistent failure retains custody. Closed contexts refuse new starts; old duties still publish. MCP init failure discharges only after method-unsent + clean transport/session join; live runners/possible sends stay unknown. Unbound fallback/legacy remain. Failed publication joins body/holds outside locks/store budgets before release. Direct Stop is cooperative, never thread kill; pooled termination/Panic stay independent.
 
 Started reviewers finish (`review_pause.py`): `review_episode` arms only at first physical handoff under Pause's lock, permitting that attempt's later rounds. The paused author leaves PENDING `in_flight`, or unsent `pending_dispatch` latched not to send; `live_detached_operations` keeps it warm until settlement. New panels refuse at $0 as `NOT_DISPATCHED`, never PASS. `finishing_reviews` names critics; Resume collects unchanged work at $0 without waiting. Process stop converts warm to exact cold pause. Unconfirmed writers keep pausing; Resume binds custody/fence/member selections. Related schedule Restore uses recorded control identity, never fan-out or adoption.
 

ouroboros/loop_tool_execution.py

@@ -1032,16 +1032,17 @@ def _execute_with_timeout(
             else:
                 submit = scope.enter_context(abandoned_on_timeout(timeout_sec, bounded=not is_reviewed_mutative))
             from ouroboros.owner_pause import OwnerPauseRefused
+            from ouroboros.store import StoreError
             try:
                 future, release_tool_custody = submit_tool_future(
                     tool_ctx, invocation["invocation_id"], fn_name, submit, _settling_tool_worker,
                     function, tools, drive_logs, live, invocation, *arguments)
-            except OwnerPauseRefused as exc:
+            except (OwnerPauseRefused, StoreError) as exc:
                 if not use_stateful:
                     raise
                 from ouroboros.tools.tool_result import launch_refusal_result
                 result = _execute_single_tool(tools, tc, drive_logs, task_id, invocation,
-                                              host_refusal=launch_refusal_result(str(exc)))
+                                              host_refusal=launch_refusal_result(exc, not_started=True))
                 return _emit_finished(tools, live, result, started_at)
             # Registration pins the caller through the wait and any late-hold
             # claim. The exact submitted future also identifies no-entry cancellation.

ouroboros/mcp_client.py

@@ -842,8 +842,11 @@ async def _call_tool_async(
     from ouroboros.owner_pause import submit_async_operation, submit_async_preparation, OwnerPauseRefused, _TOOL_OPERATION
     from ouroboros.model_wait import ModelWaitInterrupted
     from ouroboros.usage_accounting import UsageAccountingError
+    from ouroboros.store import StoreError
+    unsent_joined = False
 
     async def _do() -> ToolResult:
+        nonlocal unsent_joined
         refusal = None
         async with _transport_factory(cfg) as transport_ctx:
             streams = transport_ctx
@@ -852,15 +855,22 @@ async def _call_tool_async(
             else:
                 read, write = streams.read, streams.write  # pragma: no cover
             async with ClientSession(read, write) as session:
-                await session.initialize()
                 try:
-                    pending = submit_async_operation(None, session.call_tool, tool_name, arguments)
-                except (OwnerPauseRefused, ModelWaitInterrupted, UsageAccountingError) as exc:
-                    refusal = exc  # Unsent; first join both contexts normally.
+                    await session.initialize()
+                except BaseException as exc:
+                    # Includes timeout cancellation before tools/call. Keep the
+                    # original exception, but join both contexts before proof.
+                    refusal = exc
                 else:
-                    result = await pending
+                    try:
+                        pending = submit_async_operation(None, session.call_tool, tool_name, arguments)
+                    except (OwnerPauseRefused, ModelWaitInterrupted, UsageAccountingError, StoreError) as exc:
+                        refusal = exc
+                    else:
+                        result = await pending  # any error here may follow a send
         if refusal is not None:
             # Failed/unfinished cleanup never reaches this positive phase fact.
+            unsent_joined = True
             active = _TOOL_OPERATION.get()
             if active:
                 active[2]["settled"] = True
@@ -869,7 +879,19 @@ async def _call_tool_async(
         return _tool_result_from_call_result(result)
 
     # Launch admission precedes transport entry: a stdio server starts there.
-    return await asyncio.wait_for(submit_async_preparation(_do), timeout=timeout_sec)
+    try:
+        pending = submit_async_preparation(_do)
+    except (OwnerPauseRefused, ModelWaitInterrupted, UsageAccountingError, StoreError) as exc:
+        exc.mcp_unsent_joined = True  # no transport task was handed off
+        raise
+    try:
+        return await asyncio.wait_for(pending, timeout=timeout_sec)
+    except BaseException as exc:
+        if unsent_joined:
+            # wait_for replaces CancelledError with TimeoutError; carry proof,
+            # not exception-class inference, over that conversion.
+            exc.mcp_unsent_joined = True
+        raise
 
 
 def _stringify_call_result(result: Any) -> str:
@@ -1439,27 +1461,38 @@ class MCPManager:
             )
             if not isinstance(result, ToolResult):
                 raise TypeError("MCP transport returned a non-ToolResult outcome")
-        except asyncio.TimeoutError:
+        except asyncio.TimeoutError as exc:
+            unsent = getattr(exc, "mcp_unsent_joined", False) is True
+            if unsent:
+                confirmed()
             text = (
                 f"⚠️ MCP_TOOL_TIMEOUT: server {cfg.id!r} did not respond in {timeout}s. "
-                "The remote outcome is unknown: side effects may already have happened, "
-                "and remote cancellation is not confirmed. Use the server-specific status/read "
-                "tool, if available, to reconcile the operation before retrying."
+                + ("The tool call was not submitted; its session and transport have closed."
+                   if unsent else "The remote outcome is unknown: side effects may already have happened, "
+                   "and remote cancellation is not confirmed. Use the server-specific status/read "
+                   "tool, if available, to reconcile the operation before retrying.")
             )
-            return ToolResult(status="timeout", code="MCP_TIMEOUT", text=text)
+            return ToolResult(status="timeout", code="MCP_TIMEOUT", text=text,
+                              meta={"mcp_call_not_sent": unsent})
         except BaseException as exc:  # noqa: BLE001 - any failure is reported
             from ouroboros.owner_pause import OwnerPauseRefused, _TOOL_OPERATION
+            from ouroboros.store import StoreError
             active = _TOOL_OPERATION.get()
-            if getattr(exc, "mcp_unsent_joined", False) or (
-                    isinstance(exc, OwnerPauseRefused) and active and active[2].get("not_started")):
+            unsent = getattr(exc, "mcp_unsent_joined", False) is True or (
+                isinstance(exc, OwnerPauseRefused) and active and active[2].get("not_started") is True)
+            if unsent:
                 confirmed()  # no call, with either no transport entry or joined cleanup
+            if unsent and isinstance(exc, (OwnerPauseRefused, StoreError)):
+                from ouroboros.tools.tool_result import launch_refusal_result, _replace_tool_result
+                return _replace_tool_result(launch_refusal_result(exc),
+                                            meta_updates={"mcp_call_not_sent": True})
             body = f"⚠️ MCP_TOOL_ERROR: {type(exc).__name__}: {_redact_error_text(exc, cfg)}"
             text = _model_facing_result(cfg, tool.raw_name, body)
             return ToolResult(
                 status="error",
                 code="MCP_ERROR",
                 text=text,
-                meta={"dynamic_provider": True},
+                meta={"dynamic_provider": True, "mcp_call_not_sent": bool(unsent)},
             )
         # Joined normal return (either SDK error bit): this invocation's call is over.
         from ouroboros.owner_pause import record_mcp_call_returned

ouroboros/owner_pause.py

@@ -806,6 +806,7 @@ def tool_handoff(source: Any, name: str):
     outcome["registry_entered"] = True
     token = _TOOL_OPERATION.set((source, name, outcome))
     root_drive, root_id, task_id = _member_coordinates(source)
+    refusal = None
     try:
         handed = _HANDED_TOOL.get()
         with _CACHE_LOCK:
@@ -823,8 +824,15 @@ def tool_handoff(source: Any, name: str):
                     retain_invocation(outcome)
             retry_presend(claim_once, check=False)
         yield outcome
+    except (OwnerPauseRefused, StoreError) as exc:
+        refusal = exc
+        raise
     finally:
-        outcome["closed"] = True
+        with outcome["phase_lock"]:
+            outcome["closed"] = True
+            if refusal is not None:
+                # Snapshot with closure, not before a competing admitted start.
+                refusal.invocation_not_started = outcome.get("not_started") is True
         _TOOL_OPERATION.reset(token)
         if outcome.get("future_row") is None:
             settle_invocation(outcome)
@@ -858,13 +866,23 @@ def operation_start(source: Any = None):
     source = source if source is not None else (active[0] if active else None)
     with launch_admission(source):
         if active and active[0] is source:
-            if active[2].get("closed"):
-                raise OwnerPauseRefused("operation_already_returned")
-            active[2]["not_started"] = False
+            _mark_tool_start(active[2])
         yield
         arm_review_episode(source)  # the submission inside succeeded under this same lock
 
 
+def _mark_tool_start(outcome: dict) -> None:
+    """Order admission against closure; never hold this lock across I/O.
+
+    Called after root admission. The durable RLock remains independent: taking
+    it here could make Pause wait for another thread's custody write.
+    """
+    with outcome["phase_lock"]:
+        if outcome.get("closed"):
+            raise OwnerPauseRefused("operation_already_returned")
+        outcome["not_started"] = False
+
+
 def submit_tool(source: Any, name: str, submit: Any, function: Any, *args: Any):
     """Hand one invocation to its existing sticky executor under Pause exclusion."""
     with launch_admission(source):
@@ -921,9 +939,7 @@ def run_tool_handler(source: Any, function: Any, /, *args: Any, **kwargs: Any):
                     if reason or remaining is not None and remaining <= 0:
                         raise PhysicalDispatchInterrupted(reason or "deadline")
                     if outcome is not None:
-                        if outcome.get("closed"):
-                            raise OwnerPauseRefused("operation_already_returned")
-                        outcome["not_started"] = False
+                        _mark_tool_start(outcome)
                     arm_review_episode(source)
             # Validate the inherited caller controls, not this tool's soft wait
             # envelope. Its retry owner must use that same context. The body
@@ -968,11 +984,12 @@ def submit_async_preparation(function: Any, *args: Any, source: Any = None, **kw
     async def invoke():
         return await function(*args, **kwargs)
 
-    with launch_admission(source if source is not None else active[0] if active else None):
+    source = source if source is not None else active[0] if active else None
+    with launch_admission(source):
+        if active and active[0] is source:
+            _mark_tool_start(active[2])  # preparation can open a process/socket
         with physical_acquisition():
             task = asyncio.create_task(invoke())
-        if active:
-            active[2]["not_started"] = False  # this transport can open a process/socket
         return task
 
 

ouroboros/tool_custody.py

@@ -11,6 +11,7 @@ from contextvars import copy_context
 import logging
 import os
 import threading
+import time
 import uuid
 from typing import Any
 
@@ -43,6 +44,8 @@ def invocation_outcome(source: Any, name: str, operation_id: str = "", *, future
             "local_owner": ({"pid": os.getpid(), "process_birth": owner.answer_owner_birth,
                              "task_attempt": owner.attempt} if bound else {}),
             "memory_owned": future_owned and bound, "lock": threading.RLock(),
+            # Phase transitions never wait for the durable lock's store I/O.
+            "phase_lock": threading.Lock(),
             "admission_context": copy_context()}
 
 
@@ -101,28 +104,47 @@ def retain_invocation(outcome: dict, update=None) -> None:
 
 
 def settle_invocation(outcome: dict) -> None:
-    """Local completion cannot retire independent pending effects."""
+    """Retry only this idempotent completion within one store budget, never work.
+
+    Store acquisitions do not wait; the durable lock is released between tries.
+    A permanent refusal keeps the result and its unresolved durable custody.
+    """
     from ouroboros.task_store import DELETE_ROW, patch_handoff
     from ouroboros.utils import utc_now_iso
+    from ouroboros.store import StoreBusy, STORE_WAIT_SEC, acquisition_budget
+    from ouroboros._usage_wait import ACQUISITION_SLICE_SEC
 
-    with outcome["lock"]:
-        if not outcome.get("durable_attempted") or outcome.get("local_holds", 0):
-            return
-        if not any(outcome.get(key) is True for key in
-                   ("not_started", "settled", "local_extension_returned", "mcp_call_returned")):
-            return
-        def settle(claim):
-            if claim is None:
-                return None
-            pending = {key: value for key, value in (claim.get("external_pending") or {}).items()
-                       if key not in outcome.get("confirmed_external", set())}
-            claim = {**claim, "state": "returned", "returned_at": utc_now_iso(), "external_pending": pending}
-            return claim if claim_still_owns_effect(outcome["claim_root"], claim) else DELETE_ROW
+    end = time.monotonic() + STORE_WAIT_SEC
+    while True:
         try:
-            patch_handoff(outcome["claim_root"], outcome["claim_task"], "launch_handoffs",
-                          outcome["operation_id"], settle)
+            with outcome["lock"], acquisition_budget(0):
+                if not outcome.get("durable_attempted") or outcome.get("local_holds", 0):
+                    return
+                if not any(outcome.get(key) is True for key in
+                           ("not_started", "settled", "local_extension_returned", "mcp_call_returned")):
+                    return
+                returned_at = outcome.setdefault("returned_at", utc_now_iso())
+                def settle(claim):
+                    if claim is None:
+                        return None
+                    pending = {key: value for key, value in (claim.get("external_pending") or {}).items()
+                               if key not in outcome.get("confirmed_external", set())}
+                    claim = {**claim, "state": "returned", "returned_at": claim.get("returned_at", returned_at),
+                             "external_pending": pending}
+                    return claim if claim_still_owns_effect(outcome["claim_root"], claim) else DELETE_ROW
+                patch_handoff(outcome["claim_root"], outcome["claim_task"], "launch_handoffs",
+                              outcome["operation_id"], settle)
+            return
+        except StoreBusy as exc:
+            remaining = end - time.monotonic()
+            if exc.reason == "contention" and remaining > 0:
+                time.sleep(min(ACQUISITION_SLICE_SEC, remaining))
+                continue
+            log.warning("Invocation custody settlement retained for %s", outcome["operation_id"], exc_info=True)
+            return
         except Exception:
             log.warning("Invocation custody settlement retained for %s", outcome["operation_id"], exc_info=True)
+            return
 
 
 def _update_active_claim(update) -> None:
@@ -157,6 +179,8 @@ def external_handoff(kind: str):
                 return
             def remove(claim):
                 pending = dict(claim.get("external_pending") or {})
+                if token not in pending:
+                    return None  # confirmation cannot manufacture an absent duty
                 pending.pop(token, None)
                 return {**claim, "external_pending": pending}
             try:
@@ -164,8 +188,26 @@ def external_handoff(kind: str):
                 discharged = True
             except Exception:
                 log.warning("External custody confirmation retained for %s", token, exc_info=True)
+        if outcome.get("closed"):
+            # This receipt ends only its external duty, not a live runner or
+            # callback owned by the enclosing Future.
+            row = outcome.get("future_row")
+            if row is not None:
+                from ouroboros.budget_pause import _finish_tool_row
+                _finish_tool_row(row)
+            else:
+                settle_invocation(outcome)
     try:
-        retry_presend(lambda: retain_invocation(outcome, add), check=False)
+        def publish():
+            # Existing DB exclusion orders this new duty against settlement,
+            # without denying legitimate late updates to earlier duties.
+            with outcome["lock"]:
+                with outcome["phase_lock"]:
+                    if outcome.get("closed"):
+                        from ouroboros.owner_pause import OwnerPauseRefused
+                        raise OwnerPauseRefused("operation_already_returned")
+                retain_invocation(outcome, add)
+        retry_presend(publish, check=False)
     except BaseException:
         confirmed()  # no caller acquired this external start
         raise
@@ -188,8 +230,9 @@ def hold_invocation():
     def release():
         with outcome["lock"]:
             outcome["local_holds"] -= 1
-            if outcome.get("closed"):
-                settle_invocation(outcome)
+            closed = outcome.get("closed")
+        if closed:
+            settle_invocation(outcome)
     return release
 
 

ouroboros/tools/extension_dispatch.py

@@ -332,6 +332,9 @@ def _dispatch_extension_tool_untagged(
     except controls:
         raise
     except Exception as exc:
+        from ouroboros.store import StoreError
+        if isinstance(exc, StoreError) and handoff is not None and handoff.get("not_started") is True:
+            raise  # registry snapshots the shared positive no-entry fact
         text = f"⚠️ TOOL_ERROR ({name}): extension tool failed: {type(exc).__name__}: {exc}"
         return _extension_result("error", "EXTENSION_ERROR", text, dispatched=True)
 

ouroboros/tools/registry_core.py

@@ -1082,10 +1082,12 @@ class ToolRegistry:
                     return f"⚠️ TOOL_ERROR ({name}): {e}", None
                 except Exception as e:
                     from ouroboros.body_candidate import CandidateRefused
+                    from ouroboros.store import StoreError
 
+                    if isinstance(e, StoreError) and handoff and handoff.get("not_started") is True:
+                        raise  # the invocation owner retains its actual entry phase
                     if isinstance(e, CandidateRefused):  # e.g. a process inside the candidate without its isolation
-                        return ToolResult(status="blocked", code=e.code, text=f"⚠️ {e.code}: {e.text}",
-                                          meta={"operation_outcome": "completed_no_effect"}), None
+                        return ToolResult(status="blocked", code=e.code, text=f"⚠️ {e.code}: {e.text}"), None
                     return f"⚠️ TOOL_ERROR ({name}): {e}", None
         finally:
             if observed_skill is not None:
@@ -1112,6 +1114,7 @@ class ToolRegistry:
 
     def _execute_legacy_text(self, name: str, args: Dict[str, Any]) -> str | ToolResult:
         from ouroboros.owner_pause import OwnerPauseRefused, tool_handoff
+        from ouroboros.store import StoreError
 
         try:
             with tool_handoff(self._ctx, str(name or "")) as handoff:
@@ -1126,12 +1129,15 @@ class ToolRegistry:
         except OwnerPauseRefused as exc:
             from ouroboros.tools.tool_result import launch_refusal_result
 
-            return launch_refusal_result(str(exc))
+            return launch_refusal_result(exc)
+        except StoreError as exc:
+            from ouroboros.tools.tool_result import launch_refusal_result
+            if getattr(exc, "invocation_not_started", None) is True:
+                return launch_refusal_result(exc)
+            raise  # preserve the outer executor's post-entry error handling
 
     def _execute_admitted_text(self, name: str, args: Dict[str, Any],
                                handoff: Optional[Dict[str, Any]] = None) -> str | ToolResult:
-        if handoff is not None:
-            handoff["not_started"] = True
         name = str(name or "").strip()
         args, presence_arg_error = _presence_bound_args(self._ctx, name, args)
         if presence_arg_error:

ouroboros/tools/services.py

@@ -532,14 +532,15 @@ def _start_service(
             )
             return json.dumps(payload, ensure_ascii=False, indent=2)
         except OwnerPauseRefused as exc:
-            return _publish_tool_result(ctx, launch_refusal_result(str(exc), completed_no_effect=True))
+            return _publish_tool_result(ctx, launch_refusal_result(exc))
         except body_candidate.CandidateRefused:
             raise
         except Exception as exc:
             text = redact_known_values(f"⚠️ SERVICE_START_ERROR: executor backend failed: {type(exc).__name__}: {exc}", secret_values)
             if getattr(exc, "process_not_started", False) is True:
                 return _publish_tool_result(ctx, ToolResult(status="error", code="LEGACY_TOOL_ERROR",
-                    text=text, meta={"operation_outcome": "completed_no_effect"}))
+                    text=text, meta={"process_not_started": True,
+                                     **({"previous_service_retired": True} if existing is not None else {})}))
             return text
     task_id = str(getattr(ctx, "task_id", "") or "manual")
     log_dir = pathlib.Path(ctx.drive_root) / "services" / task_id
@@ -605,7 +606,7 @@ def _start_service(
         log_fh.close()
     except OwnerPauseRefused as exc:
         log_fh.close()
-        return _publish_tool_result(ctx, launch_refusal_result(str(exc), completed_no_effect=True))
+        return _publish_tool_result(ctx, launch_refusal_result(exc))
     except Exception as exc:
         log_fh.close()
         return redact_known_values(f"⚠️ SERVICE_START_ERROR: {type(exc).__name__}: {exc}", secret_values)

ouroboros/tools/shell.py

@@ -531,7 +531,7 @@ def _run_shell(
         result = _masked_green_disclosure(ctx, _publish_process_result(ctx, "SHELL_REGEX_AUTO_CORRECTED" if regex_autocorrected else "OK", text, exit_code=0, artifact_registered=bool(artifact_registered and not artifact_failed), shell_regex_auto_corrected=regex_autocorrected, meta=process_meta), cmd)
         return _disclose_output_audit_failure(ctx, result, audit_error)
     except OwnerPauseRefused as exc:
-        return _publish_tool_result(ctx, launch_refusal_result(str(exc), completed_no_effect=True))
+        return _publish_tool_result(ctx, launch_refusal_result(exc))
     except subprocess.TimeoutExpired:
         _publish_unfinished_process_facts(ctx, _command_start_ts, timed_out=True)
         # Timeout-created scratch still needs its exclusion fingerprint.

ouroboros/tools/tool_result.py

@@ -166,7 +166,7 @@ TOOL_CODE_SPECS: Mapping[str, ToolCodeSpec] = MappingProxyType(
             "use an authority permitted by the task contract",
         ),
         # ``body_candidate.CandidateRefused``: the task's own-body candidate could not be
-        # prepared, resumed or isolated; nothing was changed and no process started.
+        # prepared, resumed or isolated; prior invocation effects are not erased.
         **{code: _code_spec("blocked", "blocked", "warning", "inspect or repair the retained body candidate")
            for code in (
             "CANDIDATE_UNAVAILABLE", "CANDIDATE_PATH_OCCUPIED", "CANDIDATE_MISSING",
@@ -370,11 +370,10 @@ TOOL_CODE_SPECS: Mapping[str, ToolCodeSpec] = MappingProxyType(
             "warning",
             "follow the refusal text",
         ),
-        # The owner paused the tree before this call's launch handoff
-        # (ouroboros/owner_pause.py): nothing ran — distinct from a timeout or an
-        # interrupted call whose execution is unknown.
+        # Refusal codes preserve authority and phase separately. Only the
+        # positive pre-entry variant proves that this handler never ran.
         "OWNER_LAUNCH_AUTHORITY_UNAVAILABLE": _code_spec(
-            "unavailable", "unavailable", "warning", "retry when launch authority is available",
+            "unavailable", "unavailable", "warning", "follow the reported phase and reconcile prior effects before retry",
         ),
         "STOP_ACTION_CONFLICT": _code_spec(
             "blocked", "blocked", "warning", "retry the original Stop action unchanged",
@@ -385,6 +384,12 @@ TOOL_CODE_SPECS: Mapping[str, ToolCodeSpec] = MappingProxyType(
             "info",
             "issue the call again after the owner resumes the task",
         ),
+        "OWNER_OPERATION_REFUSED": _code_spec(
+            "blocked", "blocked", "warning", "reconcile prior effects before repeating the call",
+        ),
+        "TOOL_ADMISSION_UNAVAILABLE": _code_spec(
+            "unavailable", "unavailable", "warning", "inspect the reported admission cause",
+        ),
         "CAPABILITY_UNAVAILABLE": _code_spec(
             "unavailable",
             "unavailable",
@@ -753,26 +758,41 @@ def completed_local_read(handler):
     return read
 
 
-def launch_refusal_result(reason: str, *, completed_no_effect: bool = False) -> ToolResult:
-    """An unsent operation; only its producer can attest completed preparation."""
-    from ouroboros.owner_pause import NOT_STARTED_TEXT
+def launch_refusal_result(reason: str | Exception, *, not_started: bool | None = None) -> ToolResult:
+    """A refused submission does not erase earlier effects of its invocation."""
+    from ouroboros.owner_pause import LaunchBusy, NOT_STARTED_TEXT, _TOOL_OPERATION
+    from ouroboros.store import StoreError
 
-    unavailable = reason == "owner_launch_authority_unavailable"
-    code = "OWNER_LAUNCH_AUTHORITY_UNAVAILABLE" if unavailable else "OWNER_PAUSE_NOT_STARTED"
-    if unavailable:
-        text = ("⚠️ OWNER_LAUNCH_AUTHORITY_UNAVAILABLE: NOT STARTED — launch authority is temporarily "
-                "unavailable. Nothing was submitted; retry when authority is available.")
-    elif reason in {"owner_pause_authority_unreadable", "model_sleep_authority_unreadable"}:
-        text = ("⚠️ OWNER_PAUSE_NOT_STARTED: NOT STARTED — pause/sleep authority could not be read. "
-                "Nothing was submitted; repair the authority before retrying.")
-    elif reason == "operation_already_returned":
-        text = "⚠️ OWNER_PAUSE_NOT_STARTED: NOT STARTED — this operation has ended; it cannot submit more work."
-    else:
-        text = NOT_STARTED_TEXT
+    if not_started is None:
+        not_started = getattr(reason, "invocation_not_started", None)
+        active = _TOOL_OPERATION.get()
+        if not_started is None and active:
+            with active[2]["phase_lock"]:
+                not_started = active[2].get("not_started")
+    store_refusal = isinstance(reason, StoreError)
+    physical = (getattr(reason, "reason", None) or f"{type(reason).__name__}: {reason}"
+                if isinstance(reason, (LaunchBusy, StoreError)) else None)
+    semantic = "store_unavailable" if store_refusal else str(reason)
+    unavailable = store_refusal or semantic == "owner_launch_authority_unavailable"
+    code = ("TOOL_ADMISSION_UNAVAILABLE" if store_refusal else
+            "OWNER_LAUNCH_AUTHORITY_UNAVAILABLE" if unavailable else
+            "OWNER_PAUSE_NOT_STARTED" if not_started is True else "OWNER_OPERATION_REFUSED")
+    phase = ("NOT STARTED — the tool handler was not entered." if not_started is True else
+             "This submission was refused; the invocation may already have effects. "
+             "Reconcile those effects before repeating the call.")
+    if not_started is True and semantic in {"owner_pause_authority_unreadable", "model_sleep_authority_unreadable"}:
+        phase = ("NOT STARTED — pause/sleep authority could not be read. The tool handler was not entered; "
+                 "repair the authority before retrying.")
+    elif not_started is True and semantic == "operation_already_returned":
+        phase = "NOT STARTED — this invocation has ended; it cannot submit more work."
+    detail = f"{semantic}; launch reason: {physical}" if physical is not None else semantic
+    text = (NOT_STARTED_TEXT if not_started is True and semantic == "owner_pause"
+            else f"⚠️ {code}: {phase}")
     return ToolResult(status="unavailable" if unavailable else "blocked", code=code,
-                      text=f"{text} ({reason})", meta={"owner_pause_not_started": True,
-                      "control_reason": reason, **({"operation_outcome": "completed_no_effect"}
-                                                   if completed_no_effect else {})})
+                      text=f"{text} ({detail})",
+                      meta={"owner_pause_not_started": not_started is True,
+                            "control_reason": semantic,
+                            **({"launch_reason": physical} if physical is not None else {})})
 
 
 def _publish_process_result(
@@ -856,6 +876,8 @@ _EXACT_IDENTIFIER_CODES = MappingProxyType(
         "USER_FILES_PATH_BLOCKED": "USER_FILES_PATH_BLOCKED",
         "OWNER_PAUSE_NOT_STARTED": "OWNER_PAUSE_NOT_STARTED",
         "OWNER_LAUNCH_AUTHORITY_UNAVAILABLE": "OWNER_LAUNCH_AUTHORITY_UNAVAILABLE",
+        "OWNER_OPERATION_REFUSED": "OWNER_OPERATION_REFUSED",
+        "TOOL_ADMISSION_UNAVAILABLE": "TOOL_ADMISSION_UNAVAILABLE",
         "STOP_ACTION_CONFLICT": "STOP_ACTION_CONFLICT",
         "COGNITIVE_TOOL_REQUIRED": "COGNITIVE_TOOL_REQUIRED",
         "RESOURCE_CONSTRAINT_BLOCKED": "RESOURCE_CONSTRAINT_BLOCKED",

tests/test_batch4_launch_completion.py

@@ -104,7 +104,10 @@ def test_pause_during_process_preparation_prevents_process_start(tmp_path, monke
         args['name'] = 'held'
     result = registry.execute_result('start_service' if service else 'run_command', args)
     assert not marker.exists(), result
-    assert result.code == 'OWNER_PAUSE_NOT_STARTED', result
+    # The process was not launched, but its tool handler/preparation did run.
+    assert result.code == 'OWNER_OPERATION_REFUSED', result
+    assert result.meta['owner_pause_not_started'] is False
+    assert result.meta.get('operation_outcome') != 'completed_no_effect'
     assert not load_task_result(tmp_path, 'root').get('launch_handoffs')
 
 

tests/test_batch4_producer_custody.py

@@ -149,7 +149,11 @@ def test_docker_popen_no_start_is_positive_no_effect(tmp_path, monkeypatch, serv
     if service:
         args["name"] = "writer"
     result = registry.execute_result("start_service" if service else "run_command", args)
-    assert result.meta.get("operation_outcome") == "completed_no_effect", result
+    if service:
+        assert result.meta["process_not_started"] is True
+        assert "operation_outcome" not in result.meta  # a replacement may have retired its old instance
+    else:
+        assert result.meta.get("operation_outcome") == "completed_no_effect", result
     assert [kind for kind, _ in calls] == ["start"]
     assert not list(_owned(tmp_path))
     _consumers(tmp_path, registry, queue, workers, False)

tests/test_batch4_repair_boundaries.py

@@ -415,6 +415,12 @@ def test_timed_out_mcp_runner_cannot_start_after_caller_return(tmp_path, monkeyp
     monkeypatch.setattr(mcp_client, "_transport_factory", transport)
     monkeypatch.setattr(mcp_client, "ClientSession", Session)
     def timed_call(*_a):
+        join = threading.Thread.join
+        def after_entry(thread, timeout=None):
+            if thread.name == "mcp-sync-runner":
+                assert entered.wait(5)  # prove preparation began before caller timeout
+            return join(thread, timeout)
+        monkeypatch.setattr(threading.Thread, "join", after_entry)
         try:
             return mcp_client._run_async(
                 lambda: mcp_client._call_tool_async(None, "effect", {}, timeout_sec=10), join_timeout=0)
@@ -427,7 +433,8 @@ def test_timed_out_mcp_runner_cannot_start_after_caller_return(tmp_path, monkeyp
     try:
         result = asyncio.run(execute())
         assert result.status == "timeout" and entered.wait(5)
-        assert not load_task_result(tmp_path, "root").get("launch_handoffs")
+        # Unlike a closed-before-entry runner, this transport is really alive.
+        assert load_task_result(tmp_path, "root").get("launch_handoffs")
     finally:
         release.set()
         assert entered.wait(5)

tests/test_body_candidate.py

@@ -241,7 +241,9 @@ def test_candidate_environment_failure_refuses_the_process_and_never_uses_the_se
 
     for result in (started, command, executor):
         assert (result.status, result.code) == ("blocked", "CANDIDATE_ENVIRONMENT_UNAVAILABLE"), result
-        assert result.meta.get("operation_outcome") == "completed_no_effect", result
+        # A missing process is not whole-call no-effect: a service replacement
+        # may already have retired its previous instance before this refusal.
+        assert "operation_outcome" not in result.meta, result
     assert unknown.status == "error" and "registry unreadable" in unknown.text, unknown  # fails closed
     assert not (candidate / "ran.txt").exists()
     assert not list((data / "services").rglob("env-refused.executor.log"))

tests/test_mcp_timeout_outcomes.py

@@ -1,4 +1,4 @@
-"""A transport timeout leaves remote effects unknown until a server-specific read."""
+"""Timeout projections distinguish clean unsent preparation from unknown effects."""
 
 from __future__ import annotations
 
@@ -47,12 +47,15 @@ def test_timeout_disclosure_survives_registry_and_status_read(
     """Use real client wait/cancel and registry dispatch; only the peer is controlled."""
     calls = []
     effects = []
-    closed_sessions = []
+    closed_sessions, closed_transports = [], []
     first_session = True
 
     @asynccontextmanager
     async def transport(_cfg):
-        yield None, None
+        try:
+            yield None, None
+        finally:
+            closed_transports.append(True)
 
     class Session:
         def __init__(self, *_streams):
@@ -94,19 +97,25 @@ def test_timeout_disclosure_survives_registry_and_status_read(
         text = result.text
     else:
         text = registry.execute("mcp_controlled__apply", args)
-    _assert_unknown(text)
+    if timeout_phase == "initialize":
+        assert "The tool call was not submitted" in text
+        assert "session and transport have closed" in text
+    else:
+        _assert_unknown(text)
+    if projection == "typed":
+        assert result.meta.get("mcp_call_not_sent") is (timeout_phase == "initialize")
     assert "reconcile" in TOOL_CODE_SPECS["MCP_TIMEOUT"].recovery
     expected = ["op-667"] if timeout_phase == "after_effect" else []
     assert effects == expected
     assert len(calls) == len(expected)  # No implicit replay after timeout.
-    assert closed_sessions == [True]  # Local close is not a remote cancel receipt.
+    assert closed_sessions == closed_transports == [True]  # Local close is not a remote cancel receipt.
 
     status = registry.execute_result("mcp_controlled__status", args)
     assert (status.status, status.code) == ("ok", "OK")
     assert status.text.endswith(json.dumps(expected))
     assert calls[-1] == ("status", args)
     assert effects == expected
-    assert closed_sessions == [True, False]
+    assert closed_sessions == [True, False] and closed_transports == [True, True]
 
 
 def test_real_stdio_effect_before_timeout_is_reconciled_by_status_read(tmp_path, registry):

tests/test_mcp_unsent_custody.py

@@ -5,7 +5,7 @@ from types import SimpleNamespace
 
 import pytest
 
-from ouroboros import mcp_client, owner_pause
+from ouroboros import mcp_client, owner_pause, store
 from tests.test_batch4_producer_custody import _registry
 from tests.test_invocation_ownership import claims, owner, run
 from tests.test_mcp_client import _settings
@@ -14,7 +14,9 @@ pytestmark = pytest.mark.serial
 
 
 @pytest.mark.parametrize("bound", [False, True])
-@pytest.mark.parametrize("phase", ["open", "sdk_error", "pause", "cleanup_unknown", "after_send"])
+@pytest.mark.parametrize("phase", ["open", "sdk_error", "pause", "cleanup_unknown", "after_send",
+                                  "init_error", "init_timeout", "init_cleanup_error",
+                                  "store_preparation", "store_admission", "store_cleanup_error", "store_after_send"])
 def test_real_manager_keeps_only_unresolved_mcp_duty(tmp_path, monkeypatch, bound, phase):
     registry, _, _ = _registry(tmp_path, monkeypatch)
     observed = []
@@ -26,7 +28,7 @@ def test_real_manager_keeps_only_unresolved_mcp_duty(tmp_path, monkeypatch, boun
             yield None, None
         finally:
             observed.append("transport_close")
-            if phase == "cleanup_unknown":
+            if phase in {"cleanup_unknown", "init_cleanup_error", "store_cleanup_error"}:
                 raise RuntimeError("transport cleanup is unconfirmed")
 
     class Session:
@@ -38,12 +40,18 @@ def test_real_manager_keeps_only_unresolved_mcp_duty(tmp_path, monkeypatch, boun
             observed.append("session_close")
         async def initialize(self):
             observed.append("initialized")
+            if phase in {"init_error", "init_cleanup_error"}:
+                raise RuntimeError("initialization failed before tools/call")
+            if phase == "init_timeout":
+                await asyncio.Event().wait()
             if phase in {"pause", "cleanup_unknown"}:
                 owner_pause.install_fence(tmp_path, "root", request_id="mcp-before-call")
         async def call_tool(self, *_args):
             observed.append("call")
             if phase == "after_send":
                 raise owner_pause.OwnerPauseRefused("a typed error after real send is not unsent proof")
+            if phase == "store_after_send":
+                raise store.StoreBusy("after method started", reason="uninspectable")
             return SimpleNamespace(content=[SimpleNamespace(type="text", text="remote answer")],
                                    isError=phase == "sdk_error")
 
@@ -57,6 +65,13 @@ def test_real_manager_keeps_only_unresolved_mcp_duty(tmp_path, monkeypatch, boun
     manager.reconfigure(_settings({"id": "demo", "enabled": True, "transport": "streamable_http",
                                    "url": "https://unused.invalid/mcp"}))
     assert manager.refresh_server("demo")["ok"]
+    if phase in {"store_preparation", "store_admission", "store_cleanup_error"}:
+        def refuse(*_args, **_kwargs):
+            raise store.StoreBusy("admission inspection failed", reason="uninspectable")
+        monkeypatch.setattr(owner_pause, "submit_async_preparation" if phase == "store_preparation"
+                            else "submit_async_operation", refuse)
+    if phase == "init_timeout":
+        manager._tool_timeout_sec = 0.02
     monkeypatch.setattr(registry, "_mcp_name_miss", lambda _name: None)
     monkeypatch.setattr(mcp_client, "_call_mcp_tool_result", manager._call_tool_result)
     if bound:
@@ -66,15 +81,29 @@ def test_real_manager_keeps_only_unresolved_mcp_duty(tmp_path, monkeypatch, boun
         async def in_loop():
             return registry.execute_result("mcp_demo__effect", {})
         result = asyncio.run(in_loop())  # real copied-context MCP runner thread
-    assert observed.count("call") == (0 if phase in {"pause", "cleanup_unknown"} else 1)
-    assert observed[-2:] == ["session_close", "transport_close"]
+    assert observed.count("call") == (0 if phase in {
+        "pause", "cleanup_unknown", "init_error", "init_timeout", "init_cleanup_error",
+        "store_preparation", "store_admission", "store_cleanup_error"} else 1)
+    assert observed == [] if phase == "store_preparation" else observed[-2:] == ["session_close", "transport_close"]
     remaining = claims(tmp_path)
-    if phase in {"cleanup_unknown", "after_send"}:
+    if phase in {"cleanup_unknown", "after_send", "init_cleanup_error", "store_cleanup_error", "store_after_send"}:
         assert result.status == "error" and len(remaining) == 1
         assert next(iter(remaining.values()))["external_pending"]
+        assert result.meta.get("mcp_call_not_sent") is not True
     else:
         assert not remaining
-        assert result.status == ("ok" if phase == "open" else "error")
+        assert result.status == {"open": "ok", "pause": "blocked", "init_timeout": "timeout",
+                                 "store_preparation": "unavailable", "store_admission": "unavailable"}.get(phase, "error")
+        if phase in {"store_preparation", "store_admission"}:
+            assert result.code == "TOOL_ADMISSION_UNAVAILABLE"
+            assert result.meta["mcp_call_not_sent"] is True
+            assert result.meta["owner_pause_not_started"] is (phase == "store_preparation")
+            assert result.meta["launch_reason"] == "uninspectable"
+        if phase in {"init_error", "init_timeout"}:
+            assert result.meta["mcp_call_not_sent"] is True
+        if phase == "pause":
+            assert result.meta["mcp_call_not_sent"] is True
+            assert result.meta["owner_pause_not_started"] is False  # preparation already ran
         if phase in {"open", "sdk_error"}:
             assert result.meta["mcp_is_error"] is (phase == "sdk_error")
             assert "remote answer" in result.text

tests/test_operation_late_confirmation.py

@@ -0,0 +1,130 @@
+"""Consumer regression: exact external confirmation cannot release a live runner."""
+import threading
+
+import pytest
+
+from ouroboros import budget_pause, owner_pause, tool_custody
+from ouroboros.presence_continuation import _custody_blockers
+from tests.test_batch4_producer_custody import _registry
+from tests.test_invocation_ownership import claims, owner, run
+
+pytestmark = pytest.mark.serial
+
+
+@pytest.mark.parametrize("bound", [False, True])
+def test_late_confirmation_waits_for_actual_runner_hold(tmp_path, monkeypatch, bound):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    confirm_now, confirmed, finish = threading.Event(), threading.Event(), threading.Event()
+    runners, outcomes, errors, calls = [], [], [], []
+
+    def body(*_args, **_kwargs):
+        calls.append(True)
+        confirm = tool_custody.external_handoff("nested transport")
+        release_hold = tool_custody.hold_invocation()
+        outcomes.append(owner_pause._TOOL_OPERATION.get()[2])
+
+        def runner():
+            try:
+                assert confirm_now.wait(5)
+                confirm()
+                confirmed.set()
+                assert finish.wait(5)
+            except BaseException as exc:
+                errors.append(exc)
+            finally:
+                release_hold()
+
+        thread = threading.Thread(target=runner, name="held-operation-runner")
+        runners.append(thread)
+        thread.start()
+        return "caller returned; runner still owned"
+
+    registry.override_handler("knowledge_read", body)
+    try:
+        if bound:
+            with owner(registry):
+                result = run(registry)["tool_result"]
+        else:
+            result = registry.execute_result("knowledge_read", {"topic": "probe"})
+        assert result.text == "caller returned; runner still owned" and calls == [True]
+        assert outcomes[0]["closed"] and len(claims(tmp_path)) == 1
+        if bound:
+            row = outcomes[0]["future_row"]
+            assert row["future"].done() and row["holds"] == 1
+            assert not row["settled"].is_set()
+        confirm_now.set()
+        assert confirmed.wait(5) and not errors
+        assert runners[0].is_alive()
+        assert len(claims(tmp_path)) == 1, "confirmation must not retire the runner's invocation"
+        assert not next(iter(claims(tmp_path).values()))["external_pending"]
+        assert any(x["kind"] == "tool_handoff" for x in _custody_blockers(tmp_path, "root"))
+        if bound:
+            assert row["holds"] == 1 and not row["settled"].is_set()
+            assert not budget_pause.drain_local_tool_futures(registry._ctx, timeout_sec=0)["drained"]
+    finally:
+        confirm_now.set()
+        finish.set()
+        for thread in runners:
+            thread.join(5)
+            assert not thread.is_alive()
+    assert not errors and not claims(tmp_path)
+    assert not _custody_blockers(tmp_path, "root")
+    if bound:
+        assert row["holds"] == 0 and row["settled"].is_set()
+
+
+def test_async_extension_mcp_join_keeps_enclosing_live_runner(tmp_path, monkeypatch):
+    """Real extension timeout and MCP SDK unwind share one Future-owned call."""
+    import asyncio
+
+    from ouroboros import extension_process_runner, mcp_client
+    from ouroboros.tools import extension_dispatch
+    from tests.test_mcp_call_completion import _wire
+
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    method_calls = _wire(monkeypatch, timeout=10)
+    initialized, fail_init, mcp_joined, finish = [threading.Event() for _ in range(4)]
+    runners, results = [], []
+    initialize = mcp_client.ClientSession.initialize
+
+    async def failing_init(session):
+        await initialize(session)  # real SDK exchange, before tools/call
+        initialized.set()
+        assert await asyncio.to_thread(fail_init.wait, 8)
+        raise RuntimeError("controlled initialization failure")
+
+    monkeypatch.setattr(mcp_client.ClientSession, "initialize", failing_init)
+
+    async def handler(**_kwargs):
+        runners.append(threading.current_thread())
+        results.append(mcp_client.get_manager()._call_tool_result("mcp_demo__effect", {}))
+        mcp_joined.set()
+        assert finish.wait(8)
+        return "runner finished"
+
+    descriptor = {"name": "ext_probe_nested", "handler": handler, "wants_ctx": False,
+                  "content_hash": "c" * 64, "skill": "", "timeout_sec": 1}
+    monkeypatch.setattr(extension_dispatch, "_extension_dispatch_candidate", lambda *_a: (descriptor, False))
+    monkeypatch.setattr(extension_dispatch, "_generation_digest_for", lambda *_a: "d" * 64)
+    monkeypatch.setattr(extension_process_runner, "disclose_inprocess_extension_dispatch", lambda *_a, **_k: None)
+    try:
+        with owner(registry):
+            result = run(registry, name="ext_probe_nested", args={}, timeout=6)["tool_result"]
+            assert result.code == "EXTENSION_TIMEOUT" and initialized.is_set(), result.text
+            assert len(claims(tmp_path)) == 1 and runners[0].is_alive()
+            fail_init.set()
+            assert mcp_joined.wait(5)
+            assert results[0].meta["mcp_call_not_sent"] is True and not method_calls
+            assert runners[0].is_alive()
+            assert len(claims(tmp_path)) == 1
+            assert any(x["kind"] == "tool_handoff" for x in _custody_blockers(tmp_path, "root"))
+            assert not budget_pause.drain_local_tool_futures(registry._ctx, timeout_sec=0)["drained"]
+    finally:
+        fail_init.set()
+        finish.set()
+        for thread in runners:
+            thread.join(8)
+            assert not thread.is_alive()
+        mcp_client.reset_manager_for_tests()
+    assert not claims(tmp_path)
+    assert not _custody_blockers(tmp_path, "root")

tests/test_operation_refusal_truth.py

@@ -0,0 +1,311 @@
+"""Refusal truth belongs to a whole invocation, not its last nested start."""
+import asyncio
+import threading
+from contextlib import asynccontextmanager, contextmanager
+from contextvars import copy_context
+
+import pytest
+
+from ouroboros import mcp_client, owner_pause
+from tests.test_batch4_producer_custody import _registry
+from tests.test_invocation_ownership import claims, owner, run
+
+pytestmark = pytest.mark.serial
+
+
+@pytest.mark.parametrize("code,status", [("OWNER_OPERATION_REFUSED", "blocked"),
+                                        ("TOOL_ADMISSION_UNAVAILABLE", "unavailable")])
+def test_text_refusal_roundtrip_keeps_execution_failure(code, status):
+    from ouroboros.loop_tool_execution import _extract_result_metadata, _is_tool_execution_failure
+    from ouroboros.store import StoreBusy
+    from ouroboros.tools.tool_result import LegacyTextResultAdapter, launch_refusal_result
+
+    refusal = (owner_pause.OwnerPauseRefused("owner_pause") if status == "blocked" else
+               StoreBusy("admission read failed", reason="permission_denied"))
+    original = launch_refusal_result(refusal, not_started=status != "blocked")
+    assert original.code == code and original.status == status
+    for tool in ("knowledge_read", "mcp_demo__effect"):
+        restored = LegacyTextResultAdapter.from_text(tool, original.text)
+        assert restored.code == code and restored.status == status
+        assert _is_tool_execution_failure(True, original.text, fn_name=tool) is True
+        assert _extract_result_metadata(tool, original.text, True)["status"] == status
+
+
+@pytest.mark.parametrize("bound", [False, True])
+def test_post_effect_refusal_preserves_whole_call_phase(tmp_path, monkeypatch, bound):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    marker, later = tmp_path / "effect", []
+
+    def body(*_args, **_kwargs):
+        marker.write_text("already happened")
+        owner_pause.install_fence(tmp_path, "root", request_id="after-effect")
+        with owner_pause.operation_start(registry._ctx):
+            later.append(True)
+
+    registry.override_handler("knowledge_read", body)
+    if bound:
+        with owner(registry):
+            result = run(registry)["tool_result"]
+    else:
+        result = registry.execute_result("knowledge_read", {"topic": "probe"})
+    assert marker.read_text() == "already happened" and later == []
+    assert result.code == "OWNER_OPERATION_REFUSED"
+    assert result.meta["owner_pause_not_started"] is False
+    assert "Nothing ran" not in result.text
+    assert not claims(tmp_path)
+
+
+@pytest.mark.parametrize("route", ["builtin", "mcp"])
+def test_refusal_keeps_physical_cause_and_positive_no_entry(tmp_path, monkeypatch, route):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    from tests.test_mcp_call_completion import _wire
+    calls = []
+    if route == "mcp":
+        calls = _wire(monkeypatch)
+
+        def refused(*_args, **_kwargs):
+            raise owner_pause.LaunchBusy("permission_denied")
+        monkeypatch.setattr(owner_pause, "submit_async_preparation", refused)
+        name, args = "mcp_demo__effect", {}
+    else:
+        @contextmanager
+        def refused(*_args, **_kwargs):
+            raise owner_pause.LaunchBusy("permission_denied")
+            yield
+        monkeypatch.setattr(owner_pause, "launch_lock", refused)
+        registry.override_handler("knowledge_read", lambda *_args, **_kwargs: calls.append(True))
+        name, args = "knowledge_read", {"topic": "probe"}
+    try:
+        result = registry.execute_result(name, args)
+        assert not calls
+        assert "permission_denied" in result.text and "temporarily" not in result.text
+        assert result.meta["launch_reason"] == "permission_denied"
+        assert result.meta["control_reason"] == "owner_launch_authority_unavailable"
+        assert result.meta["owner_pause_not_started"] is True
+        assert not claims(tmp_path)
+    finally:
+        mcp_client.reset_manager_for_tests()
+
+
+def test_closed_copied_context_cannot_enter_mcp_transport(tmp_path, monkeypatch):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    contexts, observed = [], []
+
+    def body(*_args, **_kwargs):
+        contexts.append(copy_context())
+        return "joined"
+    registry.override_handler("knowledge_read", body)
+    assert registry.execute_result("knowledge_read", {"topic": "probe"}).text == "joined"
+
+    @asynccontextmanager
+    async def transport(_cfg):
+        observed.append("entered")
+        yield None, None
+    monkeypatch.setattr(mcp_client, "_MCP_SDK_AVAILABLE", True)
+    monkeypatch.setattr(mcp_client, "_transport_factory", transport)
+    with pytest.raises(owner_pause.OwnerPauseRefused, match="operation_already_returned"):
+        contexts[0].run(asyncio.run, mcp_client._call_tool_async(None, "probe", {}, timeout_sec=1))
+    assert observed == []
+
+
+def test_phase_admission_does_not_wait_on_durable_lock(tmp_path, monkeypatch):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    held, release, started = (threading.Event() for _ in range(3))
+    errors = []
+    with owner_pause.tool_handoff(registry._ctx, "probe") as outcome:
+        def writer():
+            with outcome["lock"]:
+                held.set()
+                assert release.wait(5)
+        thread = threading.Thread(target=writer)
+        thread.start()
+        assert held.wait(2)
+        context = copy_context()
+
+        def start():
+            try:
+                with owner_pause.operation_start(registry._ctx):
+                    started.set()
+            except BaseException as exc:
+                errors.append(exc)
+        starter = threading.Thread(target=lambda: context.run(start))
+        starter.start()
+        try:
+            assert started.wait(2), "durable I/O must not hold the phase admission"
+            # The root gate is free while the independent durable writer waits.
+            owner_pause.install_fence(tmp_path, "root", request_id="during-durable-write")
+        finally:
+            release.set()
+            starter.join(5)
+            thread.join(5)
+        assert not errors and not starter.is_alive() and not thread.is_alive()
+        outcome["settled"] = True
+
+
+@pytest.mark.parametrize("close_first", [False, True])
+def test_close_and_start_share_one_atomic_phase_boundary(tmp_path, monkeypatch, close_first):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    contexts, outcomes, submitted, errors = [], [], [], []
+    ready, release = threading.Event(), threading.Event()
+
+    def own():
+        with owner_pause.tool_handoff(registry._ctx, "probe") as outcome:
+            contexts.append(copy_context())
+            outcomes.append(outcome)
+            ready.set()
+            assert release.wait(5)
+            outcome["settled"] = True
+    thread = threading.Thread(target=own)
+    thread.start()
+    assert ready.wait(2)
+    outcome = outcomes[0]
+    if close_first:
+        release.set()
+        thread.join(5)
+    else:
+        # Keep the admitted action open while the invocation closes: admission
+        # is ordered, not a promise that all independent effects have finished.
+        outcome["phase_lock"].acquire()
+
+    def start():
+        try:
+            with owner_pause.operation_start(registry._ctx):
+                submitted.append(True)
+        except owner_pause.OwnerPauseRefused as exc:
+            errors.append(str(exc))
+    if close_first:
+        contexts[0].run(start)
+        assert submitted == [] and errors == ["operation_already_returned"]
+    else:
+        outcome["phase_lock"].release()
+        contexts[0].run(start)
+        release.set()
+        thread.join(5)
+        assert submitted == [True] and errors == []
+    assert not thread.is_alive() and outcome["closed"] is True
+
+
+@pytest.mark.parametrize("failure", ["pause", "pre_popen_error", "candidate_environment"])
+def test_retired_service_is_not_whole_call_no_effect(tmp_path, monkeypatch, failure):
+    from types import SimpleNamespace
+
+    from ouroboros.tools import services
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    ctx = registry._ctx
+    marker = tmp_path / "retired"
+    key = services._service_key(ctx, "replace")
+    monkeypatch.setitem(services._SERVICES, key, SimpleNamespace(proc=SimpleNamespace(poll=lambda: 0)))
+
+    def retire(*_args, **_kwargs):
+        marker.write_text("retired previous record")
+        return "stopped"
+    monkeypatch.setattr(services, "_stop_service", retire)
+    monkeypatch.setattr(services, "_executor_can_run_cwd", lambda *_args: True)
+
+    def refused(*_args, **_kwargs):
+        if failure == "pause":
+            raise owner_pause.OwnerPauseRefused("owner_pause")
+        if failure == "candidate_environment":
+            from ouroboros.body_candidate import CandidateRefused
+            raise CandidateRefused("CANDIDATE_ENVIRONMENT_UNAVAILABLE", "isolation unavailable")
+        error = OSError("before new process")
+        error.process_not_started = True
+        raise error
+    monkeypatch.setattr(services, "executor_start_service", refused)
+    result = registry.execute_result("start_service", {"cmd": ["ignored"], "name": "replace"})
+    assert marker.exists()
+    assert result.meta.get("operation_outcome") != "completed_no_effect"
+    assert result.meta.get("owner_pause_not_started") is not True
+    assert "Nothing ran" not in result.text
+    if failure == "pre_popen_error":
+        assert result.meta["process_not_started"] is True
+        assert result.meta["previous_service_retired"] is True
+
+
+def test_executor_retirement_is_not_denied_by_missing_host_record(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+
+    from ouroboros import body_candidate, workspace_executor
+    from ouroboros.tools import services
+
+    registry, _, _ = _registry(tmp_path, monkeypatch, backend="local")
+    ctx = registry._ctx
+    retired = []
+    key = workspace_executor.service_key(ctx, "replace")
+    old = SimpleNamespace(service_id=key)
+    monkeypatch.setattr(services, "_SERVICES", {})
+    monkeypatch.setitem(workspace_executor._SERVICES, key, old)
+    monkeypatch.setattr(workspace_executor, "_service_state", lambda _record: "exited")
+
+    def retire(record):
+        retired.append(record)
+        workspace_executor._SERVICES.pop(key)
+        return {"state": "exited"}
+
+    def refuse(*_args, **_kwargs):
+        error = OSError("environment failed before replacement Popen")
+        error.process_not_started = True
+        raise error
+
+    monkeypatch.setattr(workspace_executor, "_stop_service_record", retire)
+    monkeypatch.setattr(body_candidate, "executor_environment", refuse)
+    result = registry.execute_result("start_service", {"cmd": ["ignored"], "name": "replace"})
+    assert retired == [old]
+    assert result.meta["process_not_started"] is True
+    assert result.meta.get("previous_service_retired") is not False
+    assert result.meta.get("operation_outcome") != "completed_no_effect"
+
+
+def test_close_cannot_split_a_live_closed_check_and_start_mark(tmp_path, monkeypatch):
+    """Pause exactly after the real start's closed read, before its mark."""
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    checked, release, closing, done = (threading.Event() for _ in range(4))
+    threads, outcomes, submissions, errors = [], [], [], []
+    def own():
+        try:
+            with owner_pause.tool_handoff(registry._ctx, "probe") as outcome:
+                outcomes.append(outcome)
+                class Observed:
+                    def get(self, key, default=None):
+                        value = outcome.get(key, default)
+                        if key == "closed" and threading.current_thread().name == "phase-starter":
+                            assert not value
+                            checked.set()
+                            assert release.wait(5)
+                        return value
+                    def __getitem__(self, key):
+                        return outcome[key]
+                    def __setitem__(self, key, value):
+                        outcome[key] = value
+                context = copy_context()
+                context.run(owner_pause._TOOL_OPERATION.set, (registry._ctx, "probe", Observed()))
+                def start():
+                    try:
+                        with owner_pause.operation_start(registry._ctx):
+                            submissions.append(True)
+                    except BaseException as exc:
+                        errors.append(exc)
+                starter = threading.Thread(target=lambda: context.run(start), name="phase-starter")
+                threads.append(starter)
+                starter.start()
+                assert checked.wait(3)
+                outcome["settled"] = True
+                closing.set()
+        except BaseException as exc:
+            errors.append(exc)
+        finally:
+            done.set()
+    owner_thread = threading.Thread(target=own)
+    owner_thread.start()
+    try:
+        assert closing.wait(5)
+        assert not done.wait(.05), "closure must not overtake the checked-but-unmarked start"
+        assert outcomes[0].get("closed") is not True
+    finally:
+        release.set()
+        owner_thread.join(5)
+        for thread in threads:
+            thread.join(5)
+    assert not owner_thread.is_alive() and not any(t.is_alive() for t in threads)
+    assert errors == [] and submissions == [True]
+    assert outcomes[0]["closed"] is True and outcomes[0]["not_started"] is False

tests/test_operation_store_boundaries.py

@@ -0,0 +1,530 @@
+"""Positive no-entry, exact completion and closed-context custody consumers."""
+import sys
+from contextvars import copy_context
+
+import pytest
+
+from ouroboros import owner_pause, store, task_store, tool_custody
+from tests.test_batch4_producer_custody import _registry
+from tests.test_invocation_ownership import claims, owner, run
+
+pytestmark = pytest.mark.serial
+
+
+@pytest.mark.parametrize("after_body", [False, True])
+def test_extension_store_refusal_preserves_dispatch_provenance(tmp_path, monkeypatch, after_body):
+    from ouroboros import extension_process_runner
+    from ouroboros.tools import extension_dispatch
+
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    calls = []
+    def body(**_kwargs):
+        calls.append(True)
+        raise store.StoreBusy("extension body entered", reason="uninspectable")
+    descriptor = {"name": "ext_probe_store", "handler": body, "wants_ctx": False,
+                  "content_hash": "c" * 64, "skill": ""}
+    monkeypatch.setattr(extension_dispatch, "_extension_dispatch_candidate", lambda *_a: (descriptor, False))
+    monkeypatch.setattr(extension_dispatch, "_generation_digest_for", lambda *_a: "d" * 64)
+    monkeypatch.setattr(extension_process_runner, "disclose_inprocess_extension_dispatch", lambda *_a, **_k: None)
+    actual = owner_pause._validate_launch
+    def refuse_on_entry(source, *args, **kwargs):
+        # The actual admission runs in its captured caller context, which need
+        # not carry the worker's _TOOL_OPERATION; fail this gate, not that proxy.
+        if not after_body:
+            raise store.StoreBusy("extension admission inspection", reason="uninspectable")
+        return actual(source, *args, **kwargs)
+    monkeypatch.setattr(owner_pause, "_validate_launch", refuse_on_entry)
+    with owner(registry):
+        result = run(registry, name="ext_probe_store", args={})["tool_result"]
+    assert calls == ([True] if after_body else [])
+    if after_body:
+        assert result.code == "EXTENSION_ERROR"
+        assert result.meta["physical_dispatch"] is True
+        assert result.meta["extension_generation"] == "d" * 64
+    else:
+        assert result.code == "TOOL_ADMISSION_UNAVAILABLE"
+        assert result.meta["owner_pause_not_started"] is True
+        assert "physical_dispatch" not in result.meta and "extension_generation" not in result.meta
+
+
+@pytest.mark.parametrize("bound", [False, True])
+@pytest.mark.parametrize("after_body", [False, True])
+def test_store_refusal_preserves_actual_entry_stage(tmp_path, monkeypatch, bound, after_body):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    calls = []
+    original = owner_pause._validate_launch
+
+    def refused(*args, **kwargs):
+        raise store.StoreBusy("controlled admission read", reason="permission_denied")
+
+    def body(*_args, **_kwargs):
+        calls.append(True)
+        raise store.StoreBusy("body already entered", reason="permission_denied")
+
+    registry.override_handler("knowledge_read", body)
+    if not after_body:
+        monkeypatch.setattr(owner_pause, "_validate_launch", refused)
+    if bound:
+        with owner(registry):
+            result = run(registry)["tool_result"]
+    else:
+        result = registry.execute_result("knowledge_read", {"topic": "probe"})
+    assert calls == ([True] if after_body else [])
+    assert result.meta.get("owner_pause_not_started") is (None if after_body else True)
+    if not after_body:
+        assert result.code == "TOOL_ADMISSION_UNAVAILABLE"
+        assert result.meta["launch_reason"] == "permission_denied"
+    else:
+        assert result.status == "error" and "NOT STARTED" not in result.text
+    monkeypatch.setattr(owner_pause, "_validate_launch", original)
+
+
+@pytest.mark.parametrize("bound", [False, True])
+def test_transient_exact_settlement_retries_no_business_work(tmp_path, monkeypatch, bound):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    patch = task_store.patch_handoff
+    attempts, body_calls, confirmed_tokens = [], [], []
+
+    def contended(root, task, kind, identity, change):
+        def inspect(claim):
+            result = change(claim)
+            if kind == "launch_handoffs" and result is task_store.DELETE_ROW:
+                attempts.append(identity)
+                if len(attempts) == 1:
+                    raise store.StoreBusy("one settle contention", reason="contention")
+            return result
+        return patch(root, task, kind, identity, inspect)
+    monkeypatch.setattr(task_store, "patch_handoff", contended)
+
+    def body(*_args, **_kwargs):
+        body_calls.append(True)
+        confirm = tool_custody.external_handoff("fixture-joined-effect")
+        confirm()
+        confirmed_tokens.append(True)
+        return "preserved result"
+    registry.override_handler("knowledge_read", body)
+    if bound:
+        with owner(registry):
+            result = run(registry)["tool_result"]
+    else:
+        result = registry.execute_result("knowledge_read", {"topic": "probe"})
+    assert result.text == "preserved result" and body_calls == confirmed_tokens == [True]
+    assert len(attempts) == 2 and attempts[0] == attempts[1]
+    assert not claims(tmp_path)
+
+
+@pytest.mark.parametrize("route", ["shell", "mcp"])
+@pytest.mark.parametrize("bound", [False, True])
+def test_late_refused_handoff_cannot_recreate_claim(tmp_path, monkeypatch, route, bound):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    contexts, opened = [], []
+
+    def body(*_args, **_kwargs):
+        contexts.append(copy_context())
+        return "joined"
+    registry.override_handler("knowledge_read", body)
+    if bound:
+        with owner(registry):
+            assert run(registry)["result"] == "joined"
+    else:
+        assert registry.execute_result("knowledge_read", {"topic": "probe"}).text == "joined"
+    assert not claims(tmp_path)
+    if route == "shell":
+        from ouroboros.tools.shell_process import _tracked_subprocess_run
+        def late():
+            return _tracked_subprocess_run([sys.executable, "-c", "pass"], cwd=str(tmp_path),
+                                           stdout=-1, stderr=-1, text=True, timeout=2)
+    else:
+        from ouroboros import mcp_client
+        from tests.test_mcp_call_completion import _wire
+        opened = _wire(monkeypatch)
+        def late():
+            return mcp_client.get_manager()._call_tool_result("mcp_demo__effect", {})
+    try:
+        with pytest.raises(owner_pause.OwnerPauseRefused, match="operation_already_returned"):
+            contexts[0].run(late)
+        assert not opened and not claims(tmp_path)
+    finally:
+        if route == "mcp":
+            mcp_client.reset_manager_for_tests()
+
+
+@pytest.mark.parametrize("reason", ["permission_denied", "contention"])
+def test_unavailable_settle_preserves_result_and_unresolved_custody(tmp_path, monkeypatch, reason):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    patch = task_store.patch_handoff
+    writes = []
+    # A structural finite retry bound, shortened only in this test, not policy.
+    monkeypatch.setattr(store, "STORE_WAIT_SEC", 0.01)
+    def failed(root, task, kind, identity, change):
+        def inspect(claim):
+            result = change(claim)
+            if result is task_store.DELETE_ROW:
+                writes.append(identity)
+                raise store.StoreBusy("settlement unavailable", reason=reason)
+            return result
+        return patch(root, task, kind, identity, inspect)
+    monkeypatch.setattr(task_store, "patch_handoff", failed)
+    calls = []
+    registry.override_handler("knowledge_read", lambda *_a, **_k: calls.append(True) or "kept")
+    assert registry.execute_result("knowledge_read", {"topic": "probe"}).text == "kept"
+    assert calls == [True] and len(claims(tmp_path)) == 1
+    assert len(set(writes)) == 1
+    if reason != "contention":
+        assert len(writes) == 1
+
+
+def test_ambiguous_completion_is_idempotent_and_preserves_other_duties(tmp_path, monkeypatch):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    patch = task_store.patch_handoff
+    raised, timestamps = [], []
+    from ouroboros.task_results import load_task_result
+    with owner_pause.tool_handoff(registry._ctx, "probe") as outcome:
+        tool_custody.retain_unconfirmed_host_operation("external publisher still pending")
+        outcome["settled"] = True
+        def uncertain(root, task, kind, identity, change):
+            def inspect(claim):
+                result = change(claim)
+                if isinstance(result, dict) and result.get("state") == "returned":
+                    timestamps.append(result["returned_at"])
+                return result
+            result = patch(root, task, kind, identity, inspect)
+            if timestamps and not raised:
+                raised.append(True)
+                raise store.StoreBusy("COMMIT completed before response was lost", reason="contention")
+            return result
+        monkeypatch.setattr(task_store, "patch_handoff", uncertain)
+    claim = load_task_result(tmp_path, "root")["launch_handoffs"][outcome["operation_id"]]
+    assert claim["state"] == "returned" and claim["host_unconfirmed"]
+    assert len(timestamps) == 2 and timestamps[0] == timestamps[1]
+
+
+def test_delayed_real_mcp_runner_cannot_open_transport_after_callers_return(tmp_path, monkeypatch):
+    """A real MCPManager -> _run_async copied thread, delayed before its coroutine.
+
+    This is scheduling delay, not another tool or a fabricated returned flag.
+    The caller's finite wait ends first; preparation then encounters closure.
+    """
+    import asyncio
+    import threading
+
+    from ouroboros import mcp_client
+    from tests.test_mcp_call_completion import _wire
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    calls = _wire(monkeypatch)
+    entered, release = threading.Event(), threading.Event()
+    threads, transports = [], []
+    actual_run, actual_join, actual_transport = asyncio.run, threading.Thread.join, mcp_client._transport_factory
+    actual_wrapper = mcp_client._run_async
+
+    def delayed(coro, *args, **kwargs):
+        if threading.current_thread().name == "mcp-sync-runner":
+            threads.append(threading.current_thread())
+            entered.set()
+            assert release.wait(5)
+        return actual_run(coro, *args, **kwargs)
+    def joined(thread, timeout=None):
+        if thread.name == "mcp-sync-runner":
+            assert entered.wait(5)
+        return actual_join(thread, timeout)
+    def transport(cfg):
+        transports.append(True)
+        return actual_transport(cfg)
+    monkeypatch.setattr(asyncio, "run", delayed)
+    monkeypatch.setattr(threading.Thread, "join", joined)
+    monkeypatch.setattr(mcp_client, "_transport_factory", transport)
+    monkeypatch.setattr(mcp_client, "_run_async", lambda factory, **_kw: actual_wrapper(factory, join_timeout=0))
+    try:
+        async def invoke():
+            return registry.execute_result("mcp_demo__effect", {})
+        result = actual_run(invoke())
+        assert result.status in {"error", "timeout"}
+        assert len(claims(tmp_path)) == 1  # real runner has not yet unwound
+    finally:
+        release.set()
+        for thread in threads:
+            actual_join(thread, 5)
+            assert not thread.is_alive()
+        mcp_client.reset_manager_for_tests()
+    assert not calls and not transports
+    # Unknown was never rewritten as no-send; only the runner's positive unwind
+    # can later refine this original exact duty (not currently collected here).
+    assert len(claims(tmp_path)) == 1
+
+
+@pytest.mark.parametrize("bound", [False, True])
+def test_body_store_error_still_runs_shell_post_checks(tmp_path, monkeypatch, bound):
+    from ouroboros.tools import registry_guard_process
+
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    marker, checked = tmp_path / "body-effect", []
+    def body(*_args, _resolved_binding=None, **_kwargs):
+        marker.write_text("already executed")
+        raise store.StoreBusy("after execution", reason="permission_denied")
+    def post_checks(_registry, result, **kwargs):
+        checked.append((result, kwargs["tool_name"]))
+        return result
+    registry.override_handler("run_command", body)
+    monkeypatch.setattr(registry_guard_process, "_run_shell_post_checks", post_checks)
+    args = {"cmd": ["sh", "-c", "true"]}
+    if bound:
+        with owner(registry):
+            result = run(registry, name="run_command", args=args)["tool_result"]
+    else:
+        result = registry.execute_result("run_command", args)
+    assert marker.exists(), result.text
+    assert marker.read_text() == "already executed"
+    assert len(checked) == 1 and checked[0][1] == "run_command"
+    assert result.status == "error" and "NOT STARTED" not in result.text
+    assert result.meta.get("owner_pause_not_started") is not True
+
+
+def test_postcheck_store_error_keeps_outer_executor_failure(tmp_path, monkeypatch):
+    from ouroboros.tools import registry_guard_process
+
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    calls, checked = [], []
+
+    def body(*_args, _resolved_binding=None, **_kwargs):
+        calls.append(True)
+        return "exit_code=0"
+
+    def post_checks(_registry, result, **kwargs):
+        checked.append(kwargs["tool_name"])
+        raise store.StoreBusy("post-check unavailable", reason="permission_denied")
+
+    registry.override_handler("run_command", body)
+    monkeypatch.setattr(registry_guard_process, "_run_shell_post_checks", post_checks)
+    with owner(registry):
+        result = run(registry, name="run_command", args={"cmd": ["sh", "-c", "true"]})
+    assert calls == [True] and checked == ["run_command"]
+    assert result["tool_result"].code == "EXECUTOR_ERROR"
+    assert result["tool_result"].meta.get("owner_pause_not_started") is not True
+
+
+def test_sticky_store_admission_refusal_never_enters_executor_body(tmp_path, monkeypatch):
+    import json
+
+    from ouroboros import loop_tool_execution as execution
+
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    entered = []
+    registry.override_handler("browser_action", lambda *_a, **_k: entered.append(True) or "done")
+    def refuse(*_args, **_kwargs):
+        raise store.StoreBusy("sticky pre-entry read", reason="permission_denied")
+    monkeypatch.setattr(owner_pause, "_validate_launch", refuse)
+    executor = execution.StatefulToolExecutor()
+    call = {"id": "sticky-store", "function": {"name": "browser_action",
+            "arguments": json.dumps({"action": "snapshot"})}}
+    try:
+        result = execution._execute_with_timeout(registry, call, tmp_path / "logs", 3, "root", executor)
+    finally:
+        executor.shutdown()
+    assert entered == []
+    typed = result["tool_result"]
+    assert typed.code == "TOOL_ADMISSION_UNAVAILABLE"
+    assert typed.meta["owner_pause_not_started"] is True
+    assert typed.meta["launch_reason"] == "permission_denied"
+    assert not claims(tmp_path)
+
+
+@pytest.mark.parametrize("route", ["shell", "mcp"])
+@pytest.mark.parametrize("bound", [False, True])
+def test_close_before_external_publication_creates_no_phantom_duty(tmp_path, monkeypatch, route, bound):
+    """Split real copied-context publication before its durable lock acquisition.
+
+    The old early phase check passed before this barrier and recreated custody
+    after closure. New publication tests phase under durable exclusion instead.
+    """
+    import threading
+
+    from ouroboros import mcp_client
+    from tests.test_mcp_call_completion import _wire
+
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    waiting, release = threading.Event(), threading.Event()
+    threads, errors, writes, starts = [], [], [], []
+    patch = task_store.patch_handoff
+    if route == "mcp":
+        starts = _wire(monkeypatch)
+    else:
+        import subprocess
+        actual_popen = subprocess.Popen
+        def unexpected_spawn(*args, **kwargs):
+            if threading.current_thread().name == "late-publisher":
+                starts.append(True)
+                raise AssertionError("closed publisher spawned a process")
+            return actual_popen(*args, **kwargs)  # owner birth observations are not the effect
+        monkeypatch.setattr(subprocess, "Popen", unexpected_spawn)
+
+    def observe(root, task, kind, identity, update):
+        def change(current):
+            result = update(current)
+            if threading.current_thread().name == "late-publisher" and isinstance(result, dict):
+                writes.append(result)
+            return result
+        return patch(root, task, kind, identity, change)
+    monkeypatch.setattr(task_store, "patch_handoff", observe)
+
+    def body(*_args, **_kwargs):
+        outcome = owner_pause._TOOL_OPERATION.get()[2]
+        actual_lock = outcome["lock"]
+        class DelayedLock:
+            delayed = False
+            def __enter__(self):
+                if threading.current_thread().name == "late-publisher" and not self.delayed:
+                    self.delayed = True
+                    waiting.set()
+                    assert release.wait(5)
+                return actual_lock.__enter__()
+            def __exit__(self, *args):
+                return actual_lock.__exit__(*args)
+        outcome["lock"] = DelayedLock()
+        context = copy_context()
+        def late():
+            try:
+                if route == "shell":
+                    from ouroboros.tools.shell_process import _tracked_subprocess_run
+                    _tracked_subprocess_run([sys.executable, "-c", "pass"], cwd=str(tmp_path), timeout=2)
+                else:
+                    mcp_client.get_manager()._call_tool_result("mcp_demo__effect", {})
+            except BaseException as exc:
+                errors.append(exc)
+        thread = threading.Thread(target=lambda: context.run(late), name="late-publisher")
+        threads.append(thread)
+        thread.start()
+        assert waiting.wait(3)
+        return "joined before late publication"
+    registry.override_handler("knowledge_read", body)
+    try:
+        if bound:
+            with owner(registry):
+                result = run(registry)["tool_result"]
+        else:
+            result = registry.execute_result("knowledge_read", {"topic": "probe"})
+        assert result.text == "joined before late publication"
+        assert not claims(tmp_path)
+    finally:
+        release.set()
+        for thread in threads:
+            thread.join(5)
+            assert not thread.is_alive()
+        mcp_client.reset_manager_for_tests()
+    assert len(errors) == 1 and isinstance(errors[0], owner_pause.OwnerPauseRefused)
+    assert not starts and not writes and not claims(tmp_path)
+
+
+def test_preclose_external_duty_survives_return_and_exact_late_confirmation(tmp_path, monkeypatch):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    confirmations, contexts = [], []
+    def body(*_args, **_kwargs):
+        confirmations.append(tool_custody.external_handoff("already admitted effect"))
+        contexts.append(copy_context())
+        return "local return"
+    registry.override_handler("knowledge_read", body)
+    assert registry.execute_result("knowledge_read", {"topic": "probe"}).text == "local return"
+    before = claims(tmp_path)
+    assert len(before) == 1
+    assert next(iter(before.values()))["state"] == "returned"
+    assert next(iter(before.values()))["external_pending"]
+    # Legitimate late updates to old custody are not a new start.
+    contexts[0].run(tool_custody.retain_unconfirmed_host_operation, "pending host receipt")
+    confirmations[0]()
+    after = claims(tmp_path)
+    assert len(after) == 1 and next(iter(after.values()))["host_unconfirmed"]
+    assert not next(iter(after.values()))["external_pending"]
+    confirmations[0]()
+    assert claims(tmp_path) == after
+
+
+def test_confirmation_after_refused_publication_does_not_manufacture_claim(tmp_path, monkeypatch):
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    patch = task_store.patch_handoff
+    def refuse(root, task, kind, identity, update):
+        def change(current):
+            result = update(current)
+            if isinstance(result, dict) and result.get("external_pending"):
+                raise store.StoreBusy("new duty never published", reason="permission_denied")
+            return result
+        return patch(root, task, kind, identity, change)
+    monkeypatch.setattr(task_store, "patch_handoff", refuse)
+    def body(*_args, **_kwargs):
+        with pytest.raises(store.StoreBusy):
+            tool_custody.external_handoff("unpublished duty")
+        return "kept"
+    registry.override_handler("knowledge_read", body)
+    with owner(registry):
+        assert run(registry)["result"] == "kept"
+    assert not claims(tmp_path)
+
+
+def test_settlement_does_not_retire_live_model_receiver(tmp_path, monkeypatch):
+    import threading
+
+    from ouroboros import usage_accounting as ua
+    from ouroboros.model_wait import TaskModelWait, operation_wait_scope
+    from supervisor.continuation_admission import conflicting_writers
+    from tests._usage_store_testing import ledger_rows
+
+    registry, queue, _ = _registry(tmp_path, monkeypatch)
+    monkeypatch.setenv("TOTAL_BUDGET", "100")
+    entered, release = threading.Event(), threading.Event()
+    consumer = TaskModelWait(task={"id": "root"}, drive_root=tmp_path,
+                             event_queue=None, worker_slot_held=True)
+    errors = []
+    def send():
+        entered.set()
+        assert release.wait(10)
+        raise RuntimeError("controlled unknown model response")
+    def receiver():
+        try:
+            with operation_wait_scope(consumer):
+                ua.execute_physical_attempt(ua.AttemptRequest(
+                    model="m", provider="test", drive_root=tmp_path,
+                    task_id="root", root_task_id="root", reservation_usd=0.5), send)
+        except RuntimeError as exc:
+            errors.append(str(exc))
+    thread = threading.Thread(target=receiver)
+    thread.start()
+    try:
+        assert entered.wait(5)
+        before = ledger_rows(tmp_path)
+        assert before[-1]["state"] == "dispatched"
+        registry.override_handler("knowledge_read", lambda *_a, **_k: "local call joined")
+        assert registry.execute_result("knowledge_read", {"topic": "x"}).text == "local call joined"
+        assert not claims(tmp_path) and not consumer.closed and thread.is_alive()
+        assert ledger_rows(tmp_path) == before
+        assert any(row["kind"] == "model_handoff" for row in conflicting_writers(queue, "root"))
+    finally:
+        release.set()
+        thread.join(10)
+        consumer.close()
+    assert not thread.is_alive() and errors == ["controlled unknown model response"]
+
+
+def test_presence_keeps_live_sibling_when_other_call_settles(tmp_path, monkeypatch):
+    import threading
+
+    from ouroboros.presence_continuation import _custody_blockers
+
+    registry, _, _ = _registry(tmp_path, monkeypatch)
+    entered, release = threading.Event(), threading.Event()
+    def sibling(*_args, **_kwargs):
+        entered.set()
+        assert release.wait(10)
+        return "sibling joined"
+    registry.override_handler("knowledge_read", sibling)
+    thread = threading.Thread(target=registry.execute_result, args=("knowledge_read", {"topic": "x"}))
+    thread.start()
+    try:
+        assert entered.wait(5)
+        before = claims(tmp_path)
+        assert len(before) == 1
+        registry.override_handler("knowledge_list", lambda *_a, **_k: "other joined")
+        assert registry.execute_result("knowledge_list", {}).text == "other joined"
+        assert claims(tmp_path) == before
+        assert any(row["kind"] == "tool_handoff" for row in _custody_blockers(tmp_path, "root"))
+    finally:
+        release.set()
+        thread.join(10)
+    assert not thread.is_alive() and not _custody_blockers(tmp_path, "root")

tests/test_tool_classification_differential.py

@@ -399,6 +399,9 @@ CURRENT_PRODUCER_CONTRACTS = {
     "MANAGED_UPDATE_POSTCOMMIT_CHANGED": (False, "ok"),
     "native:LEGACY_TOOL_ERROR:MANAGED_UPDATE_POSTCOMMIT_CHANGED": (True, "error"),
     "OWNER_LAUNCH_AUTHORITY_UNAVAILABLE": (True, "unavailable"),
+    # Direction06 adds producer-owned whole-invocation phase, not a rewritten historical golden.
+    "OWNER_OPERATION_REFUSED": (True, "blocked"),
+    "TOOL_ADMISSION_UNAVAILABLE": (True, "unavailable"),
     "STOP_ACTION_CONFLICT": (True, "blocked"),
     "SAFETY_ADVICE": (False, "ok"),
     "LIGHT_MODE_REPO_CHANGED": (False, "ok"),

tests/test_tool_handoff_contention.py

@@ -148,9 +148,17 @@ def test_hard_or_unknown_claim_failure_never_retries_or_enters_handler(reader, m
 
     monkeypatch.setattr(task_store, "exists", refuse)
     monkeypatch.setattr(queue_observation, "_sleep", lambda _: pytest.fail("hard claim retried"))
-    with pytest.raises(type(error)) as caught:
-        registry.execute_result("read_file", {"path": "answer.txt"})
-    assert caught.value is error and attempts == [True] and calls == []
+    if isinstance(error, store.StoreError):
+        result = registry.execute_result("read_file", {"path": "answer.txt"})
+        assert result.code == "TOOL_ADMISSION_UNAVAILABLE"
+        assert result.meta["owner_pause_not_started"] is True
+        physical = getattr(error, "reason", None) or f"{type(error).__name__}: {error}"
+        assert result.meta["launch_reason"] == physical and physical in result.text
+    else:
+        with pytest.raises(type(error)) as caught:
+            registry.execute_result("read_file", {"path": "answer.txt"})
+        assert caught.value is error
+    assert attempts == [True] and calls == []
 
 
 def test_unconfirmed_rollback_does_not_replay_claim(reader, monkeypatch):

tests/tool_classification_corpus.py

@@ -310,7 +310,11 @@ _PRODUCER_SHAPES = (
     # tools/registry_core.py::ToolRegistry._execute_legacy_text (owner Batch4): the
     # owner paused the tree before this call's launch handoff; nothing ran.
     ("owner_pause_not_started", "knowledge_read", "⚠️ OWNER_PAUSE_NOT_STARTED: NOT STARTED — the owner paused this task tree before this operation was launched. Nothing ran; it may be issued again after Resume. (owner_pause)", "OWNER_PAUSE_NOT_STARTED", (("owner_pause_not_started", True),)),
+    # Historical emitted wording stays in the legacy corpus; current producers
+    # preserve phase/cause separately and are exercised by live round-trip tests.
     ("owner_launch_authority_unavailable", "knowledge_read", "⚠️ OWNER_LAUNCH_AUTHORITY_UNAVAILABLE: NOT STARTED — launch authority is temporarily unavailable. Nothing was submitted; retry when authority is available.", "OWNER_LAUNCH_AUTHORITY_UNAVAILABLE", (("owner_pause_not_started", True),)),
+    ("owner_operation_refused", "knowledge_read", "⚠️ OWNER_OPERATION_REFUSED: This submission was refused; the invocation may already have effects. Reconcile those effects before repeating the call. (owner_pause)", "OWNER_OPERATION_REFUSED", (("owner_pause_not_started", False),)),
+    ("tool_admission_unavailable", "knowledge_read", "⚠️ TOOL_ADMISSION_UNAVAILABLE: NOT STARTED — the tool handler was not entered. (store_unavailable; launch reason: permission_denied)", "TOOL_ADMISSION_UNAVAILABLE", (("owner_pause_not_started", True), ("launch_reason", "permission_denied"))),
     ("stop_action_conflict", "cancel_task", "⚠️ STOP_ACTION_CONFLICT: stop_action_conflict; nothing was changed.", "STOP_ACTION_CONFLICT", ()),
     ("resource_policy", "read_file", "⚠️ RESOURCE_POLICY_BLOCKED: task_contract.resource_policy protects 'blackbox'.", "RESOURCE_POLICY_BLOCKED", ()),
     ("cognitive_redirect", "write_file", "⚠️ COGNITIVE_TOOL_REQUIRED: cognitive memory is not written via 'write_file'.", "COGNITIVE_TOOL_REQUIRED", ()),