Preserve native live execution through explicit maintenance pause

README.md

@@ -57,6 +57,9 @@ approval policy, transport authentication, persistence location, and shutdown.
 Runtime history is bounded observation, not a replay log. Saved interrupted jobs
 are evidence and are never automatically dispatched again.
 
+[Cooperative execution suspension](docs/execution-suspension.md) is a separate
+opt-in path for acknowledged safe boundaries and explicit once-only resumption.
+
 If a provider, hook, or host guard cancels an active turn, the live owner settles
 through the same cancellation cleanup as Stop. It emits `generation.detached`
 with cancellation unconfirmed and `session.closed` with cancelled status; it

amplifier_module_loop_live/job_store.py

@@ -68,18 +68,34 @@ def _write(self, row):
                 os.unlink(temporary)
         self.rows[row["call_id"]] = row
 
-    def begin(self, call, job_id, receipt, native_call=None):
+    def begin(self, call, job_id, receipt, native_call=None, *, generation=None, parallel_group_id=None):
         if call.id in self.rows:
             raise RuntimeError("Saved tool call identity cannot be dispatched again")
         self._write({"version": 1, "job_id": job_id, "call_id": call.id, "tool_call": call.model_dump(),
-                     "native_call": native_call, "receipt": receipt, "status": "pending"})
+                     "native_call": native_call, "receipt": receipt, "status": "pending",
+                     "parallel_group_id": parallel_group_id,
+                     "generation": copy.deepcopy(generation)})
 
     def finish(self, call_id, result, outcome):
         row = self.rows[call_id]
         if row["status"] != "pending":
             raise RuntimeError("Saved job already has a terminal report")
         self._write({**row, "status": outcome, "result": serialize_result(result, outcome)})
 
+    def suspend(self, call_id, continuation):
+        row = self.rows[call_id]
+        if row["status"] != "pending":
+            raise RuntimeError("Only a pending job can suspend")
+        self._write({**row, "status": "paused", "continuation": copy.deepcopy(continuation)})
+
+    def resume_paused(self, call_id, continuation):
+        row = self.rows[call_id]
+        if row["status"] != "paused" or row.get("continuation") != continuation:
+            raise RuntimeError("Job continuation changed or was already consumed")
+        # Consume durably before invocation. A crash afterward is uncertain,
+        # never a reason to replay the paused continuation.
+        self._write({**row, "status": "pending", "continuation": None})
+
     def recover(self, transcript):
         """Reconcile saved evidence without redispatching work.
 
@@ -91,6 +107,11 @@ def recover(self, transcript):
         recovered = []
         for identity, original in list(self.rows.items()):
             row = original
+            if row["status"] == "paused":
+                # A host must explicitly admit this exact continuation. Keep
+                # its original receipt; neither synthesize completion nor
+                # invoke the tool while restoring saved history.
+                continue
             changed = False
             if row["status"] == "pending":
                 row = {**row, "status": "interrupted", "result": json.dumps({

amplifier_module_loop_live/runtime.py

@@ -69,6 +69,40 @@ def __init__(self, session_id=None, observer=None, max_input_chars=16000):
         self.steering_wake = None
         self.steering_closed = False
         self.stop_requested = False
+        self.pause_epoch = None
+        self.suspended = None
+        self.pause_origin = None
+        self.pause_terminal = None
+        self.owned_jobs = {}
+        self.pause_jobs = {}
+        self.execution_suspension_support = None
+
+    async def request_pause(self, epoch):
+        """Hold admission immediately; the engine proves its settled boundary."""
+        if not isinstance(epoch, str) or not epoch or len(epoch) > 128:
+            raise ValueError("A bounded maintenance epoch is required")
+        if self.execution_suspension_support is False:
+            raise RuntimeError("The native provider owns execution and does not advertise safe-boundary suspension")
+        if self.pause_epoch not in {None, epoch}:
+            raise RuntimeError("A different maintenance epoch already owns this runtime")
+        if self.pause_epoch is None:
+            self.pause_origin = copy.deepcopy(self.generation)
+            self.pause_terminal = None
+            self.pause_jobs = copy.deepcopy(self.owned_jobs)
+        self.pause_epoch = epoch
+        self.inbox.put_nowait(("maintenance_pause", epoch))
+        await self.emit("maintenance.requested", epoch=epoch)
+
+    async def release_paused(self, epoch):
+        if self.pause_epoch != epoch:
+            raise RuntimeError("Maintenance epoch changed")
+        self.inbox.put_nowait(("maintenance_release", epoch))
+
+    async def resume_paused(self, epoch):
+        """Host-authorized intent; the loop still checks each continuation owner."""
+        if self.pause_epoch != epoch:
+            raise RuntimeError("Maintenance epoch changed")
+        self.inbox.put_nowait(("maintenance_resume", epoch))
 
     async def submit(self, command: Input):
         if self.closed:
@@ -92,6 +126,8 @@ async def submit(self, command: Input):
             if previous != command:
                 raise ValueError("Command identity reused with different content")
             return command.id
+        if self.pause_epoch and command.kind in {"user", "steer", "service"}:
+            raise RuntimeError("New work is held for maintenance")
         if command.target_generation_id is not None:
             if command.kind != "steer" or not isinstance(command.target_generation_id, str) or not 1 <= len(command.target_generation_id) <= 128:
                 raise ValueError("A bounded generation anchor is only valid for steering")
@@ -148,6 +184,19 @@ async def emit(self, event_type, **data):
             self.steering_closed = False
             self.generation = {"id": data["generation_id"], "initial_input_id": data.get("initial_input_id"), "input_ids": [], "accepted_input_ids": []}
         generation = self.generation
+        if event_type == "job.queued":
+            self.owned_jobs[data["job_id"]] = {
+                "job_id": data["job_id"], "call_id": data["call_id"], "status": "pending",
+                "generation": {"id": data.get("generation_id"),
+                               "input_ids": list(data.get("input_ids", []))}}
+        elif event_type in {"job.returned", "job.failed", "job.cancelled"}:
+            job_id = data.get("job_id")
+            if job_id in self.owned_jobs:
+                self.owned_jobs[job_id]["status"] = event_type.split(".")[1]
+            if job_id in self.pause_jobs:
+                self.pause_jobs[job_id]["status"] = event_type.split(".")[1]
+        if event_type == "generation.started" and data.get("result_origin"):
+            self.owned_jobs.pop(data["result_origin"]["jobId"], None)
         if generation is not None:
             if event_type in {"input.delivered", "steering.applied"}:
                 identity = data.get("input_id")
@@ -182,6 +231,8 @@ async def emit(self, event_type, **data):
         async with self.changed:
             self.changed.notify_all()
         if event_type in {"generation.finished", "generation.failed", "generation.detached"}:
+            if self.pause_epoch and self.generation:
+                self.pause_terminal = copy.deepcopy(self.generation)
             self.generation = None
         return event
 

docs/execution-suspension.md

@@ -0,0 +1,57 @@
+# Live execution suspension
+
+Maintenance is opt-in and uses loop-streaming's public versioned continuation
+interface. The application owns the host-wide hold and exact authorization; the
+loop owns accepted input, generations, jobs and original completion attribution.
+It does not introduce another scheduler.
+
+The host installs loop-streaming's `execution.suspension` and `execution.resume`
+capabilities, plus:
+
+- `live.suspend(envelope)`: asynchronously persist the exact private envelope and
+  matching canonical history, returning `True` only after durability.
+- `live.suspended_execution`: the saved envelope when remounting explicitly.
+- `live.maintenance_admit(epoch)`: asynchronously validate current host authority
+  immediately before resuming owned work.
+
+`Runtime.request_pause(epoch)` closes new input admission immediately. Existing
+identified receipt replays remain readable. The loop waits for started operations
+to settle; cooperating delegates can suspend through their original tool owner.
+Once all owned execution reaches a boundary, the loop records version 1 and emits
+`session.suspended`. `release_paused(epoch)` checkpoints and exits with status
+`paused`, not completion/cancellation. Release never escalates to process killing.
+Provider-native opaque execution advertises no support and refuses this request.
+
+The envelope preserves the original generation, accepted/pending/steering input,
+root continuation, exact background job/call/generation origins, and separately
+attributed completed results waiting for delivery. A delegate that finishes while
+pausing retains its original result input; it is never dispatched again. Persistent
+workers retain their lifetime owner even after their current turn finishes.
+
+A fresh host remounts unchanged canonical history and the envelope without normal
+interruption repair rewriting it. It still validates history, configuration and
+current authority before dispatch. `resume_paused(epoch)` consumes only the
+approved continuation and paused jobs. The original delegate's `resume_execution`
+owns child result completion. Saved stopped/claimed/unknown work is not replayed.
+Ordinary crash recovery without an acknowledged pause remains evidence-only.
+
+Durable suspension of background jobs additionally requires `live.jobs`. The
+loop refuses to acknowledge a checkpoint with paused background jobs unless its
+ledger contains the matching job/call/continuation records. Fresh remount rejects
+missing, extra, or inconsistent paused ledger jobs (including their generation)
+before restoring runtime state. Paused ledger rows without an outer envelope
+also refuse mounting; they remain held for host inspection, not redispatch.
+Ledger-free ordinary execution and suspension with no background jobs remain
+supported. The outer envelope must have version 1 and a valid maintenance epoch;
+the host still owns binding that private envelope to the correct session.
+
+New background ledger records preserve `parallel_group_id` through resumption.
+Older records without that optional field retain `None`; recovery does not invent
+an original correlation identity. Shared maintenance admission authorizes the
+set of owned work; each engine/tool still enforces its own continuation claim.
+`session.resumed` reports that handoff, not successful completion of every claim.
+
+Hosts must distinguish waiting, acknowledged suspension, unsupported engines and
+uncertain persistence/hand-off. An arbitrary hanging tool cannot safely be
+cancelled by this protocol. Separate emergency process stopping can lose remote
+outcome information and is not a resumable checkpoint.

tests/test_continuation_guard.py

@@ -41,7 +41,7 @@ async def evaluate(*args):
 async def test_opt_in_guard_allows_normal_continuation(monkeypatch):
     loop = BundleLiveOrchestrator({'goal_stall_threshold': 100})
     guard = AsyncMock(return_value=True)
-    coordinator = SimpleNamespace(session_state={'goal': {'turns_used': 0, 'last_reason': None, 'condition': 'finish task', 'cap': 5}}, cancellation=SimpleNamespace(is_cancelled=False), get_capability=lambda name: guard)
+    coordinator = SimpleNamespace(session_state={'goal': {'turns_used': 0, 'last_reason': None, 'condition': 'finish task', 'cap': 5}}, cancellation=SimpleNamespace(is_cancelled=False), get_capability=lambda name: guard if name == 'live.continuation_guard' else None)
     actual_turn = AsyncMock(side_effect=['First response', 'Final response'])
     monkeypatch.setattr(StreamingOrchestrator, '_execute_one_turn', actual_turn)
     loop._evaluate_goal = AsyncMock(side_effect=[(False, 'Continue'), (True, 'Evidence verified')])

tests/test_loop.py

@@ -433,7 +433,7 @@ async def modify(event, data):
 
     async def test_dispatch_does_not_run_when_intent_cannot_be_saved(self):
         await self.start()
-        ledger = SimpleNamespace(begin=lambda *args: (_ for _ in ()).throw(OSError("disk unavailable")))
+        ledger = SimpleNamespace(begin=lambda *args, **kwargs: (_ for _ in ()).throw(OSError("disk unavailable")))
         self.session.coordinator.register_capability("live.jobs", ledger)
         with self.assertRaises(OSError):
             await self.loop._execute_tool_only(ToolCall(id="no-save", name="delegate", arguments={}),

tests/test_maintenance_suspension.py

@@ -0,0 +1,155 @@
+"""Real Core sessions suspend/release/remount without replaying user input."""
+
+import asyncio
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from amplifier_core import AmplifierSession, ToolResult
+from amplifier_core.message_models import ChatResponse, TextBlock, ToolCall
+
+from amplifier_module_loop_live.runtime import Input, Runtime
+
+
+class Provider:
+    name = "fixture"
+    config = {}
+
+    def __init__(self):
+        self.requests, self.responses = asyncio.Queue(), asyncio.Queue()
+
+    def get_info(self):
+        return SimpleNamespace(default_model="fixture")
+
+    async def complete(self, request, **kwargs):
+        await self.requests.put(request)
+        return await self.responses.get()
+
+    def parse_tool_calls(self, response):
+        return response.tool_calls or []
+
+    async def request(self):
+        return await asyncio.wait_for(self.requests.get(), 3)
+
+    async def reply(self, text="", calls=None):
+        await self.responses.put(ChatResponse(content=[TextBlock(text=text)], tool_calls=calls))
+
+
+class Effect:
+    name = "effect"
+    description = "A counted deterministic effect"
+    input_schema = {"type": "object", "properties": {}}
+
+    def __init__(self):
+        self.count = 0
+
+    async def execute(self, arguments):
+        self.count += 1
+        return ToolResult(success=True, output="effect completed")
+
+
+class Owner:
+    def __init__(self, root, runtime, context):
+        self.root, self.runtime, self.context = Path(root), runtime, context
+        self.root.mkdir(exist_ok=True)
+
+    def requested(self):
+        return self.runtime.pause_epoch is not None
+
+    async def checkpoint(self, frame):
+        (self.root / "boundary.json").write_text(json.dumps({
+            "continuation": frame, "messages": await self.context.get_messages()
+        }))
+        return True
+
+    async def live_checkpoint(self, envelope):
+        (self.root / "live.json").write_text(json.dumps(envelope))
+        return True
+
+    async def claim(self, frame):
+        claimed = self.root / ("claimed-" + frame["id"])
+        if claimed.exists() or self.requested() or self.runtime.stop_requested:
+            return False
+        claimed.write_text("consumed")
+        return True
+
+    async def admit(self, epoch):
+        return self.runtime.pause_epoch == epoch and not self.runtime.stop_requested
+
+
+async def start(root, provider, effect, *, restored=False):
+    runtime = Runtime(session_id="maintenance-root")
+    session = AmplifierSession({"session": {
+        "orchestrator": {"module": "loop-live", "config": {"min_delay_between_calls_ms": 0}},
+        "context": {"module": "context-simple"}}, "providers": []},
+        session_id=runtime.session_id)
+    await session.initialize()
+    coordinator = session.coordinator
+    await coordinator.mount("providers", provider, name="fixture")
+    await coordinator.mount("tools", effect, name="effect")
+    context = coordinator.get("context")
+    owner = Owner(root, runtime, context)
+    coordinator.register_capability("live.runtime", runtime)
+    coordinator.register_capability("execution.suspension", owner)
+    coordinator.register_capability("execution.resume", owner.claim)
+    coordinator.register_capability("live.suspend", owner.live_checkpoint)
+    coordinator.register_capability("live.maintenance_admit", owner.admit)
+    if restored:
+        boundary = json.loads((Path(root) / "boundary.json").read_text())
+        await context.set_messages(boundary["messages"])
+        coordinator.register_capability("live.suspended_execution",
+                                        json.loads((Path(root) / "live.json").read_text()))
+    task = asyncio.create_task(session.execute(""))
+    await runtime.wait_for(lambda e: e["type"] == "session.ready", 3)
+    return session, runtime, task
+
+
+@pytest.mark.asyncio
+async def test_pause_release_fresh_core_remount_explicit_resume_same_generation(tmp_path):
+    provider, effect = Provider(), Effect()
+    session, runtime, task = await start(tmp_path, provider, effect)
+    try:
+        await runtime.submit(Input("user", "original request", id="original-input"))
+        await provider.request()
+        generation = runtime.generation["id"]
+        await runtime.request_pause("maintenance-epoch")
+        await provider.reply(calls=[ToolCall(id="effect-call", name="effect", arguments={})])
+        await runtime.wait_for(lambda e: e["type"] == "session.suspended", 3)
+        assert effect.count == 0 and provider.requests.empty()
+        assert runtime.generation["id"] == generation
+        with pytest.raises(RuntimeError, match="held"):
+            await runtime.submit(Input("user", "old browser new send"))
+        await runtime.release_paused("maintenance-epoch")
+        await asyncio.wait_for(task, 3)
+        assert runtime.events[-1]["type"] == "session.closed"
+        assert runtime.events[-1]["status"] == "paused"
+        assert not any(e["type"] in {"generation.finished", "generation.failed", "generation.detached"}
+                       for e in runtime.events)
+    finally:
+        if not task.done():
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+        await session.cleanup()
+
+    session, runtime, task = await start(tmp_path, provider, effect, restored=True)
+    try:
+        await runtime.wait_for(lambda e: e["type"] == "session.suspended", 3)
+        assert effect.count == 0 and provider.requests.empty()
+        await runtime.resume_paused("maintenance-epoch")
+        request = await provider.request()
+        assert effect.count == 1
+        assert str(request.messages).count("original request") == 1
+        await provider.reply("finished")
+        finished = await runtime.wait_for(lambda e: e["type"] == "generation.finished", 3)
+        assert finished["generation_id"] == generation
+        assert finished["input_ids"] == ["original-input"]
+        assert not any(e["type"] == "generation.started" for e in runtime.events)
+        with pytest.raises(RuntimeError, match="changed"):
+            await runtime.resume_paused("maintenance-epoch")
+        assert effect.count == 1
+    finally:
+        await runtime.submit(Input("stop"))
+        await asyncio.wait_for(task, 3)
+        await session.cleanup()