Add cooperative safe-boundary execution suspension

README.md

@@ -2,6 +2,9 @@
 
 Token-level streaming orchestration for real-time response delivery.
 
+Hosts can opt into [cooperative execution suspension](docs/execution-suspension.md)
+at settled model/tool boundaries, with explicit durable checkpoint and resume ownership.
+
 ## Prerequisites
 
 - **Python 3.11+**

amplifier_module_loop_streaming/dispatch.py

@@ -7,12 +7,14 @@
 
 import asyncio
 import json
+import logging
 from contextlib import contextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass
 from uuid import uuid4
 
 from amplifier_core.message_models import ToolCall
+from .suspension import NestedExecutionHeld, ResumingTool, ToolExecutionSuspended
 
 _RUN = ContextVar("approved_dispatch_run", default=None)
 _CALL = ContextVar("approved_dispatch_call", default=None)
@@ -210,6 +212,21 @@ def still_authorized(child_scope):
                     self.scope.parallel_group_id or self.scope.call.id,
                     coordinator,
                 )
+        except ToolExecutionSuspended as suspended:
+            # The nested invocation is not the enclosing controller. Propagating
+            # its "not_started" frame would replay a controller that may already
+            # have performed effects. Controllers have no portable stack here.
+            try:
+                await self.scope.hooks.emit("tool:dispatch_suspended", {
+                    "source": source, "outcome": "held", "effects": "not_rolled_back",
+                })
+            except Exception as error:
+                logging.getLogger(__name__).warning(
+                    "Nested suspension observer failed: %s", type(error).__name__)
+            raise NestedExecutionHeld(
+                source, "Nested tool suspended; the enclosing controller already started. "
+                "Inspect retained effects; do not replay the controller."
+            ) from suspended
         except asyncio.CancelledError:
             await self.scope.hooks.emit(
                 "tool:dispatch_cancelled",
@@ -325,6 +342,20 @@ def reset_model_budget(self):
         if run and run.owner is self:
             run.model_calls = 0
 
+    def _budget_run(self):
+        run = _RUN.get()
+        if run is None or run.owner is not self or not run.active:
+            raise RuntimeError("Nested inference budget requires an active dispatch run")
+        return run
+
+    def model_budget_used(self):
+        return self._budget_run().model_calls
+
+    def restore_model_budget(self, used):
+        if type(used) is not int or used < 0:
+            raise ValueError("Invalid nested inference continuation budget")
+        self._budget_run().model_calls = used
+
     @contextmanager
     def run_scope(self, coordinator):
         existing = _RUN.get()
@@ -467,6 +498,9 @@ async def complete(self, request):
                                 "Nested inference exceeds provider input budget"
                             )
                 lease._valid()
+                suspension = coordinator.get_capability("execution.suspension")
+                if suspension is not None and suspension.requested():
+                    raise NestedExecutionHeld(source, "Nested inference is held for execution suspension")
                 scope.run.model_calls += 1
                 self.owner._llm_calls_this_turn += 1
                 pending = asyncio.create_task(provider.complete(request, **options))
@@ -526,7 +560,13 @@ def wrap_tools(self, tools, observed_name=None, observation=None, authority=None
             ):
                 wrapped[name] = raw
                 continue
-            lock = self.locks.setdefault(id(raw), asyncio.Lock())
+            # Continuations are per call; execution serialization is per mounted
+            # tool. Unwrap only our known adapters for identity, not execution:
+            # replacing raw would replay execute() instead of resume_execution().
+            lock_owner = raw
+            while isinstance(lock_owner, (GuardedTool, ResumingTool)):
+                lock_owner = lock_owner.tool
+            lock = self.locks.setdefault(id(lock_owner), asyncio.Lock())
             wrapped[name] = GuardedTool(
                 raw,
                 lock,

amplifier_module_loop_streaming/suspension.py

@@ -0,0 +1,104 @@
+"""Public, versioned continuation at settled execution boundaries.
+
+The host owns durable storage and admission. The engine owns this small program
+counter. A continuation never contains a Python frame, provider connection, or
+an already-running tool invocation.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any
+
+
+class ExecutionSuspended(BaseException):
+    """Execution has a host-acknowledged checkpoint, not a completed result."""
+
+    def __init__(self, continuation: dict[str, Any]):
+        super().__init__("Execution suspended at a durable boundary")
+        self.continuation = continuation
+
+
+class ToolExecutionSuspended(BaseException):
+    """A cooperating tool stopped with an explicit public continuation.
+
+    Only the tool's ``resume_execution`` method can continue this operation.
+    Calling ``execute`` again is never a valid fallback.
+    """
+
+    def __init__(self, continuation: dict[str, Any]):
+        super().__init__("Tool execution suspended")
+        self.continuation = canonical_copy(continuation)
+
+
+class ResumingTool:
+    execution_resumption = True
+
+    def __init__(self, tool, continuation):
+        self.tool, self.continuation = tool, continuation
+        if not callable(getattr(tool, "resume_execution", None)):
+            raise ValueError("Tool does not support explicit execution resumption")
+
+    def __getattr__(self, name):
+        return getattr(self.tool, name)
+
+    async def execute(self, arguments):
+        return await self.tool.resume_execution(self.continuation)
+
+
+class NestedExecutionHeld(RuntimeError):
+    """An active controller cannot be reconstructed from its nested boundary."""
+
+    def __init__(self, source, message):
+        super().__init__(message)
+        self.source = canonical_copy(source)
+
+    def tool_result(self):
+        from amplifier_core import ToolResult
+        return ToolResult(
+            success=False,
+            output={"status": "held", "effects": "not_rolled_back",
+                    "retryable": False, "message": str(self), "source": self.source},
+            error={"message": str(self), "code": "nested_execution_held", "retryable": False},
+        )
+
+
+def canonical_copy(value: Any) -> Any:
+    """Fail rather than silently discard nonportable continuation state."""
+    return json.loads(json.dumps(value, sort_keys=True, allow_nan=False))
+
+
+def fingerprint(value: Any) -> str:
+    return hashlib.sha256(
+        json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
+    ).hexdigest()
+
+
+def validate(continuation: dict[str, Any], messages: list, config: dict) -> dict:
+    frame = canonical_copy(continuation)
+    if frame.get("version") != 1 or frame.get("engine") != "loop-streaming":
+        raise ValueError("Unsupported execution continuation")
+    if frame.get("phase") not in {"provider", "prepared_request", "tools", "settling_tools", "goal", "prepared_finalization"}:
+        raise ValueError("Unsupported execution continuation phase")
+    if frame.get("history") != fingerprint(messages):
+        raise ValueError("Execution continuation history changed; inspect before resuming")
+    if frame.get("configuration") != fingerprint(config):
+        raise ValueError("Execution configuration changed; inspect before resuming")
+    if not isinstance(frame.get("iteration"), int) or frame["iteration"] < 0:
+        raise ValueError("Invalid execution continuation iteration")
+    model_calls = frame.get("dispatch_model_calls", 0)
+    if type(model_calls) is not int or model_calls < 0:
+        raise ValueError("Invalid nested inference continuation budget")
+    if frame["phase"] in {"tools", "settling_tools"} and not frame.get("tool_calls"):
+        raise ValueError("Tool continuation has no pending calls")
+    if frame["phase"] == "settling_tools" and (
+        len(frame.get("settled_tools", [])) != len(frame["tool_calls"])
+    ):
+        raise ValueError("Tool continuation outcomes do not match calls")
+    if frame["phase"] in {"prepared_request", "prepared_finalization"} and not isinstance(frame.get("prepared_request"), dict):
+        raise ValueError("Prepared continuation has no request")
+    if frame["phase"] == "goal" and (not isinstance(frame.get("goal_loop"), dict) or
+            frame["goal_loop"].get("phase") not in {"evaluate", "evaluated", "judge", "judged", "finish", "next_turn"}):
+        raise ValueError("Unknown goal continuation phase")
+    return frame

docs/execution-suspension.md

@@ -0,0 +1,62 @@
+# Cooperative execution suspension
+
+This opt-in public interface separates suspension from cancellation and normal
+completion. With no `execution.suspension` capability, ordinary execution is
+unchanged. A host supplying the capability owns durable storage, admission, and
+the explicit decision to resume. The engine never chooses a storage path or
+automatically resumes after process loss.
+
+The capability object supplies:
+
+- `requested() -> bool`: a synchronous, side-effect-free hold check.
+- `async checkpoint(frame) -> True`: retain the frame and matching canonical
+  history durably, or raise/refuse. Only `True` acknowledges persistence.
+
+The engine waits for an in-flight model response or ordinary tool invocation to
+settle, then checkpoints before another model call or tool effect. Concurrent
+tool batches settle every started sibling, retain ordered outcomes, and preserve
+unstarted calls. It does not cancel an arbitrary tool and call that a safe pause.
+A hung external operation can remain waiting indefinitely.
+
+`ExecutionSuspended` is a `BaseException` carrying the acknowledged frame. It is
+not a completed response or a failed tool result. Hosts handle this outcome
+separately and release resources only after canonical checkpoint acknowledgment.
+
+Mount the same configuration with the saved canonical messages and call
+`await orchestrator.resume(frame, context, providers, tools, hooks, coordinator)`.
+Register `execution.resume` as an async one-time admission callback accepting that
+exact frame. History/configuration, cancellation, goal state and callback
+authorization are checked before effects. A used or uncertain claim is never
+retried implicitly. The host rechecks current permission, budget, task and
+ownership constraints; old configuration is not permission to bypass controls.
+
+Frame version 1 contains explicit engine program state, not a coroutine stack.
+It records the phase (`provider`, `prepared_request`, `tools`, `settling_tools`,
+`prepared_finalization`, or `goal`), original iteration/accounting, ordered tool
+outcomes and history/configuration fingerprints. Prepared requests retain
+completed hook/context/rate-limit preparation. Goal assessment, judgment and
+continuation state retain obtained results and accounting. Frames are private
+execution state and can contain full model requests and tool outputs.
+
+A cooperating long-lived tool can raise `ToolExecutionSuspended` with its own
+versioned continuation. Its mounted tool implements
+`async resume_execution(continuation)`; the engine calls that original completion
+owner instead of `execute(arguments)`. There is no redispatch fallback. The tool
+owns child/session/result attribution and timeout/admission checks.
+
+Resumed batches validate every required completion owner before scheduling any
+sibling. A refusal after host admission does not release or reuse that claim.
+An unexpected ordinary dispatch failure during a resumed batch is an unknown,
+non-retryable outcome; started siblings still settle and retain their known
+results. Cancellation remains distinct from this failure path.
+
+True-stream input-overflow recovery checks for suspension after recovery,
+preflight and rate-limit waits, before sending the replacement request. Its
+prepared checkpoint retains that exact request, spent one-shot recovery
+allowance and accounting. Authorized resume does not repeat preparation or grant
+another recovery attempt. Any provider chunk, including a non-text chunk, still
+forbids overflow recovery.
+
+This interface does not undo remote effects, make opaque provider-native
+execution suspendable, or authorize serialization of custom private component
+state. Hosts must report unsupported or unconfirmed cases honestly.

docs/programmatic-capabilities.md

@@ -40,3 +40,25 @@ zero disables them). It is separate from the existing foreground iteration cap,
 not a claim that routing, goal evaluation, decision endpoints and financial
 budgets are already governed by a single meter. A broader inference service can
 be extracted later without putting that policy in a UI controller.
+
+Cooperative suspension expires the old run/call leases. Explicit resume starts
+new scopes while retaining the turn's spent nested-inference allowance; it does
+not grant another allowance or make the old lease usable. A hold observed after
+nested inference policy/budget preparation refuses the send before provider use.
+
+A running controller has no serialized Python stack. If one of its nested tool
+calls suspends, dispatch fails that nested call explicitly rather than treating
+the enclosing controller as unstarted. Earlier effects remain evidence and the
+controller must not be replayed. Ordinary controllers settle before the engine
+acknowledges its next safe suspension boundary.
+
+An uncaught nested hold produces a failed tool result with `status: held`,
+`effects: not_rolled_back`, nested source attribution, and `retryable: false`.
+Nested tool holds emit `tool:dispatch_suspended`; nested inference refusals use
+the existing attributed `provider:error` path. These are outcome classifications,
+not durable controller checkpoints or an authorization to retry.
+
+Nested-budget persistence applies only while Streaming owns its dispatcher.
+Missing or expired owned run scopes fail budget snapshot/restore explicitly.
+Legacy loop-live dispatchers retain their separate capability contract; they
+cannot borrow Streaming's inference capability without its active scope.