From 14167ad47462540f8ab9ba8b05356d311e326d0e Mon Sep 17 00:00:00 2001
From: Brian Krabach <brkrabac@microsoft.com>
Date: Tue, 20 Oct 2026 12:54:44 -0700
Subject: [PATCH 1/7] Retain explicit live input and delegate-job continuations
 during maintenance

---
 amplifier_module_loop_live/job_store.py    |  24 ++-
 amplifier_module_loop_live/orchestrator.py | 229 ++++++++++++++++-----
 amplifier_module_loop_live/runtime.py      |  32 +++
 tests/test_maintenance_suspension.py       | 155 ++++++++++++++
 4 files changed, 392 insertions(+), 48 deletions(-)
 create mode 100644 tests/test_maintenance_suspension.py

diff --git a/amplifier_module_loop_live/job_store.py b/amplifier_module_loop_live/job_store.py
index 85cfa06..5956ef3 100644
--- a/amplifier_module_loop_live/job_store.py
+++ b/amplifier_module_loop_live/job_store.py
@@ -68,11 +68,12 @@ 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):
         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",
+                     "generation": copy.deepcopy(generation)})
 
     def finish(self, call_id, result, outcome):
         row = self.rows[call_id]
@@ -80,6 +81,20 @@ def finish(self, call_id, result, outcome):
             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 +106,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({
diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py
index 9a5bc17..b2bdbef 100644
--- a/amplifier_module_loop_live/orchestrator.py
+++ b/amplifier_module_loop_live/orchestrator.py
@@ -14,6 +14,7 @@
 
 from amplifier_core import HookResult
 from amplifier_module_loop_streaming import StreamingOrchestrator, ConversationProviderPin
+from amplifier_module_loop_streaming.suspension import ExecutionSuspended, ToolExecutionSuspended, ResumingTool
 from .runtime import Input, observation
 from .provenance import InputContext, ContinuationContext, input_metadata
 from .job_store import serialize_result
@@ -37,11 +38,13 @@ def __init__(self, config=None):
         self.root_provider = None
         self.approved_dispatch = ApprovedDispatch(self)
 
-    async def _execute_guarded_goal(self, prompt, context, providers, tools, hooks, coordinator):
+    async def _execute_guarded_goal(self, prompt, context, providers, tools, hooks, coordinator, resume=None):
         # An explicit input always receives its first turn. Only automatic goal
         # continuations consult the host at the last safe admission boundary.
         token = _GOAL_SCOPE.set({"coordinator": coordinator, "turns": 0, "result": ""})
         try:
+            if resume is not None:
+                return await super().resume(resume, context, providers, tools, hooks, coordinator)
             return await super().execute(prompt, context, providers, tools, hooks, coordinator)
         finally:
             _GOAL_SCOPE.reset(token)
@@ -151,9 +154,24 @@ async def _execute_live(self, prompt, context, providers, tools, hooks, coordina
         if runtime is None:
             # CLI's production spawner expects a finite result from workers.
             try:
-                result=await self._execute_guarded_goal(prompt, context, providers, tools, hooks, coordinator)
+                result=await self._execute_guarded_goal(prompt, context, providers, tools, hooks, coordinator,
+                    resume=coordinator.get_capability("execution.restored") if coordinator else None)
                 self.host.finite_finished(finite_scope, coordinator, "completed", result)
                 return result
+            except ExecutionSuspended as paused:
+                # Native Core's execute ABI returns text and may translate all
+                # Python exceptions into execution errors. A host which owns
+                # the finite child can accept this explicit control outcome,
+                # rather than mistaking an exception string for proof of pause.
+                suspended = getattr(self.host, "finite_suspended", None)
+                if suspended:
+                    accepted = suspended(finite_scope, coordinator, paused.continuation)
+                    if __import__("inspect").isawaitable(accepted):
+                        accepted = await accepted
+                    if accepted is True:
+                        return ""
+                self.host.finite_finished(finite_scope, coordinator, "paused")
+                raise
             except BaseException as exc:
                 self.host.finite_finished(finite_scope, coordinator, "cancelled" if isinstance(exc,asyncio.CancelledError) else "error")
                 raise
@@ -218,7 +236,7 @@ async def observe(event, data):
         unregister = [hooks.register(event, observe, name="live-manager-" + event)
                       for event in ("content_block:end", "llm:stream_block_delta", "tool:pre", "tool:post", "tool:error")]
 
-        async def turn(command, *, continuation=False):
+        async def turn(command, *, continuation=False, resume=None):
             activation = coordinator.get_capability("live.activation")
             activation_token = None
             owner_token = LIVE_OWNER.set(self)
@@ -226,9 +244,10 @@ async def turn(command, *, continuation=False):
             try:
                 if activation:
                     activation_token = activation.bind(command.activation)
-                await self._synchronize_job_results(context)
+                if resume is None:
+                    await self._synchronize_job_results(context)
                 turn_context = context
-                if continuation:
+                if continuation or resume is not None:
                     turn_context = ContinuationContext(context)
                 else:
                     await runtime.emit("input.delivered", input_id=command.id,
@@ -238,8 +257,11 @@ async def turn(command, *, continuation=False):
                     turn_context = InputContext(turn_context, self._text(command), command)
                 stage = "manager_turn"
                 result = await self._execute_guarded_goal(
-                    "" if continuation else self._text(command), turn_context, providers, tools, hooks, coordinator)
+                    "" if continuation or resume is not None else self._text(command),
+                    turn_context, providers, tools, hooks, coordinator, resume=resume)
                 await runtime.inbox.put(("bundle_turn", (result, None)))
+            except ExecutionSuspended as paused:
+                await runtime.inbox.put(("bundle_suspended", (command, paused.continuation)))
             except asyncio.CancelledError:
                 raise
             except Exception as exc:
@@ -249,6 +271,43 @@ async def turn(command, *, continuation=False):
                     activation.reset(activation_token)
                 LIVE_OWNER.reset(owner_token)
 
+        restored = coordinator.get_capability("live.suspended_execution")
+        def restored_input(row):
+            return Input(**{**row, "attachments": tuple(row.get("attachments", ()))})
+        if restored:
+            runtime.pause_epoch = restored["epoch"]
+            runtime.pause_origin = restored.get("pause_origin")
+            runtime.pause_terminal = restored.get("pause_terminal")
+            runtime.generation = restored.get("generation")
+            runtime.suspended = restored.get("turn")
+            self.pending.extend(restored_input(row) for row in restored.get("pending", []))
+            runtime.accepted.update({row["id"]: restored_input(row) for row in restored.get("accepted", [])})
+            runtime.pending_steering.extend(restored_input(row) for row in restored.get("steering", []))
+            runtime.steering_outcomes.update(restored.get("steering_outcomes", {}))
+            runtime.steering_closed = restored.get("steering_closed", False)
+
+        async def paused_checkpoint():
+            save = coordinator.get_capability("live.suspend")
+            if not callable(save):
+                raise RuntimeError("This host cannot preserve a live suspension")
+            def command_data(command):
+                return {key: value for key, value in vars(command).items() if key != "activation"}
+            envelope = {"version": 1, "epoch": runtime.pause_epoch,
+                        "pause_origin": runtime.pause_origin, "pause_terminal": runtime.pause_terminal,
+                        "generation": runtime.generation, "turn": runtime.suspended,
+                        "pending": [command_data(row) for row in self.pending],
+                        "accepted": [command_data(row) for row in runtime.accepted.values()],
+                        "steering": [command_data(row) for row in runtime.pending_steering],
+                        "steering_outcomes": runtime.steering_outcomes,
+                        "steering_closed": runtime.steering_closed,
+                        "jobs": [{"job_id": identity, "call_id": job["call_id"],
+                                  "status": job.get("status"), "continuation": job.get("continuation"),
+                                  "generation": job.get("generation")}
+                                 for identity, job in self.jobs.items() if job.get("result") is None]}
+            if await save(envelope) is not True:
+                raise RuntimeError("Live suspension checkpoint was not acknowledged")
+            return envelope
+
         try:
             await runtime.emit("session.ready", mode="configured_bundle", native=bool(native),
                 steering="native" if native else "request_boundary", async_tools="native" if native else "delegate_job_handles" if
@@ -261,9 +320,9 @@ async def turn(command, *, continuation=False):
             if prompt:
                 self.pending.append(Input("user", prompt))
             while True:
-                if active is None and self.pending:
+                if active is None and self.pending and not runtime.pause_epoch:
                     await self._hold_anchored("generation-ended")
-                if active is None and self.pending and runtime.inbox.empty():
+                if active is None and self.pending and runtime.inbox.empty() and not runtime.pause_epoch:
                     command = self.pending.popleft()
                     # Publish busy state before scheduling: a host control can
                     # otherwise park after dequeue but before turn() first runs.
@@ -271,7 +330,7 @@ async def turn(command, *, continuation=False):
                                        initial_input_id=command.id, call_id=command.call_id)
                     active = asyncio.create_task(turn(command))
                     idle = False
-                if active is None and not self.pending and not self._active_jobs() and runtime.inbox.empty() and not idle:
+                if active is None and not self.pending and not self._active_jobs() and runtime.inbox.empty() and not idle and not runtime.pause_epoch:
                     idle = True
                     await runtime.emit("session.idle", text=last)
                     # A host may relinquish its durable session writer here while
@@ -282,6 +341,12 @@ async def turn(command, *, continuation=False):
                     park = coordinator.get_capability("live.park")
                     if park:
                         await park(activation=last_activation)
+                if (runtime.pause_epoch and active is None and not self._active_jobs()
+                        and runtime.inbox.empty() and not idle):
+                    await paused_checkpoint()
+                    await runtime.emit("session.suspended", epoch=runtime.pause_epoch,
+                                       generation_id=(runtime.generation or {}).get("id"))
+                    idle = True
                 event = await runtime.inbox.get()
                 kind, value = event
                 activation = coordinator.get_capability("live.activation")
@@ -292,6 +357,57 @@ async def turn(command, *, continuation=False):
                     captured = getattr(event, "activation", None)
                     activation.bind(captured)
                     last_activation = captured
+                if kind == "maintenance_pause":
+                    idle = False
+                    continue
+                if kind == "maintenance_release":
+                    if value != runtime.pause_epoch or active is not None or self._active_jobs():
+                        await runtime.emit("maintenance.release_refused", epoch=value, reason="execution-not-settled")
+                        continue
+                    await paused_checkpoint()
+                    status = "paused"
+                    break
+                if kind == "maintenance_resume":
+                    admit = coordinator.get_capability("live.maintenance_admit")
+                    if (value != runtime.pause_epoch or not callable(admit)
+                            or runtime.stop_requested or coordinator.cancellation.is_cancelled
+                            or await admit(value) is not True
+                            or runtime.stop_requested or coordinator.cancellation.is_cancelled):
+                        await runtime.emit("maintenance.resume_refused", epoch=value)
+                        continue
+                    # The durable host admission is authoritative. A resumed
+                    # turn keeps its generation and original input identity.
+                    saved = runtime.suspended
+                    runtime.pause_epoch, runtime.suspended = None, None
+                    if saved:
+                        command = restored_input(saved["command"])
+                        if activation:
+                            command = __import__("dataclasses").replace(command, activation=activation.current())
+                        active = asyncio.create_task(turn(command, resume=saved["continuation"]))
+                    for job_id, job in self.jobs.items():
+                        if job.get("status") == "paused":
+                            from amplifier_core.message_models import ToolCall
+                            call = ToolCall(**job["tool_call"])
+                            continuation = job["continuation"]
+                            if ledger:
+                                ledger.resume_paused(call.id, continuation)
+                            job["status"] = "pending"
+                            job["task"] = asyncio.create_task(self._run_background_job(
+                                job_id, call, tools, hooks, None, coordinator, continuation=continuation))
+                    idle = False
+                    await runtime.emit("session.resumed", epoch=value)
+                    continue
+                if kind == "bundle_suspended":
+                    await active
+                    command, frame = value
+                    active = None
+                    runtime.suspended = {"command": {key: item for key, item in vars(command).items()
+                                                     if key != "activation"}, "continuation": frame}
+                    idle = False
+                    continue
+                if kind == "bundle_job_suspended":
+                    idle = False
+                    continue
                 if kind == "input":
                     runtime.queued_inputs -= 1
                     if value.activation is not None:
@@ -346,7 +462,7 @@ async def turn(command, *, continuation=False):
                     # never replay the previous prompt or completed tools.
                     pending = [command for command in [*self.pending, *runtime.pending_steering]
                                if command.target_generation_id == runtime.generation['id']]
-                    if (pending and not runtime.stop_requested
+                    if (pending and not runtime.pause_epoch and not runtime.stop_requested
                             and not coordinator.cancellation.is_cancelled and not self._budget_exhausted):
                         active = asyncio.create_task(turn(pending[0], continuation=True))
                         continue
@@ -389,7 +505,8 @@ async def turn(command, *, continuation=False):
             if active:
                 active.cancel()
                 await asyncio.gather(active, return_exceptions=True)
-            await self._hold_anchored("session-closing")
+            if status != "paused":
+                await self._hold_anchored("session-closing")
             if active:
                 await runtime.emit("generation.detached", cancellation="unconfirmed")
             for job in self.jobs.values():
@@ -423,7 +540,8 @@ async def turn(command, *, continuation=False):
             activation = coordinator.get_capability("live.activation")
             if checkpoint and (not activation or activation.current_valid):
                 try:
-                    await self._synchronize_job_results(context)
+                    if status != "paused":
+                        await self._synchronize_job_results(context)
                     await asyncio.wait_for(checkpoint(status), 2)
                 except Exception as exc:
                     await runtime.emit("persistence.failed", error_type=type(exc).__name__)
@@ -472,45 +590,64 @@ async def _execute_tool_scoped(self, tool_call, tools, hooks, parallel_group_id,
             from .scope import NATIVE_REQUEST
             provider = NATIVE_REQUEST.get()
             ledger.begin(tool_call, job_id, receipt,
-                         provider.async_calls.get(tool_call.id) if provider else None)
-
-        async def work():
-            from .scope import JOB_CALL
-            job_token=JOB_CALL.set(tool_call.id)
-            # Isolate turn counters and ephemeral hook buffers from the manager.
-            # The original engine still runs tool:pre/approval, dispatch identity,
-            # cancellation tracking, tool:post and output modification.
-            executor = StreamingOrchestrator(self.config)
-            injections = ()
-            try:
-                job = self.jobs[job_id]
-                job["started"] = True
-                if job.get("cancel_requested"):
-                    raise asyncio.CancelledError
-                _, _, result = await executor._execute_tool_only(
-                    tool_call, tools, hooks, parallel_group_id, coordinator)
-                outcome, injections = "returned", executor._pending_ephemeral_injections
-            except asyncio.CancelledError:
-                result, outcome = None, "cancelled"
-            except Exception as exc:
-                result, outcome = type(exc).__name__, "failed"
-            finally:
-                JOB_CALL.reset(job_token)
-            # Commit after the normal tool post-hooks, before the inbox/event
-            # publication. A crash in that gap must preserve the actual report.
+                         provider.async_calls.get(tool_call.id) if provider else None,
+                         generation=runtime.generation)
+
+
+        self.jobs[job_id] = {"task": asyncio.create_task(self._run_background_job(job_id, tool_call, tools, hooks, parallel_group_id, coordinator)), "call_id": tool_call.id,
+                             "tool_call": tool_call.model_dump(), "receipt": receipt, "status": "pending",
+                             "generation": __import__("copy").deepcopy(runtime.generation)}
+        self._tool_calls_this_turn += 1
+        origin = self.jobs[job_id]["generation"] or {}
+        await runtime.emit("job.queued", job_id=job_id, call_id=tool_call.id, tool=tool_call.name,
+                           generation_id=origin.get("id"), input_ids=list(origin.get("input_ids", [])))
+        return tool_call.id, tool_call.name, self.jobs[job_id]["receipt"]
+
+    async def _run_background_job(self, job_id, tool_call, tools, hooks, parallel_group_id, coordinator, continuation=None):
+        runtime = self.runtime
+        ledger = coordinator.get_capability("live.jobs") if coordinator else None
+        from .scope import JOB_CALL
+        job_token=JOB_CALL.set(tool_call.id)
+        # Isolate turn counters and ephemeral hook buffers from the manager.
+        # The original engine still runs tool:pre/approval, dispatch identity,
+        # cancellation tracking, tool:post and output modification.
+        executor = StreamingOrchestrator(self.config)
+        injections = ()
+        try:
+            job = self.jobs[job_id]
+            job["started"] = True
+            if job.get("cancel_requested"):
+                raise asyncio.CancelledError
+            mounted = tools if continuation is None or continuation.get("kind") == "not_started" else {
+                **tools, tool_call.name: ResumingTool(tools.get(tool_call.name), continuation)}
+            _, _, result = await executor._execute_tool_only(
+                tool_call, mounted, hooks, parallel_group_id, coordinator)
+            outcome, injections = "returned", executor._pending_ephemeral_injections
+        except ToolExecutionSuspended as paused:
             try:
                 if ledger:
-                    ledger.finish(tool_call.id, result, outcome)
+                    ledger.suspend(tool_call.id, paused.continuation)
             except Exception as exc:
                 await runtime.inbox.put(("bundle_persistence_error", type(exc).__name__))
                 return
-            await runtime.inbox.put(("bundle_job", (job_id, result, outcome, injections)))
-
-        self.jobs[job_id] = {"task": asyncio.create_task(work()), "call_id": tool_call.id,
-                             "tool_call": tool_call.model_dump(), "receipt": receipt, "status": "pending"}
-        self._tool_calls_this_turn += 1
-        await runtime.emit("job.queued", job_id=job_id, call_id=tool_call.id, tool=tool_call.name)
-        return tool_call.id, tool_call.name, self.jobs[job_id]["receipt"]
+            job.update(status="paused", continuation=paused.continuation)
+            await runtime.inbox.put(("bundle_job_suspended", job_id))
+            return
+        except asyncio.CancelledError:
+            result, outcome = None, "cancelled"
+        except Exception as exc:
+            result, outcome = type(exc).__name__, "failed"
+        finally:
+            JOB_CALL.reset(job_token)
+        # Commit after the normal tool post-hooks, before the inbox/event
+        # publication. A crash in that gap must preserve the actual report.
+        try:
+            if ledger:
+                ledger.finish(tool_call.id, result, outcome)
+        except Exception as exc:
+            await runtime.inbox.put(("bundle_persistence_error", type(exc).__name__))
+            return
+        await runtime.inbox.put(("bundle_job", (job_id, result, outcome, injections)))
 
     async def _job_returned(self, job_id, result, outcome, injections=(), *, deliver):
         job = self.jobs[job_id]
diff --git a/amplifier_module_loop_live/runtime.py b/amplifier_module_loop_live/runtime.py
index becaf76..f3cd5cf 100644
--- a/amplifier_module_loop_live/runtime.py
+++ b/amplifier_module_loop_live/runtime.py
@@ -69,6 +69,34 @@ 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
+
+    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.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_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 +120,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"}:
+            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")
@@ -182,6 +212,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
 
diff --git a/tests/test_maintenance_suspension.py b/tests/test_maintenance_suspension.py
new file mode 100644
index 0000000..496e4b3
--- /dev/null
+++ b/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()

From 2e2de0dfc80a7611057f691a3bd26db0d079e126 Mon Sep 17 00:00:00 2001
From: Brian Krabach <brkrabac@microsoft.com>
Date: Tue, 20 Oct 2026 13:48:40 -0700
Subject: [PATCH 2/7] Preserve terminal job results and current activation
 through explicit resume

---
 README.md                                  |  3 ++
 amplifier_module_loop_live/orchestrator.py | 32 +++++++++++++++--
 amplifier_module_loop_live/runtime.py      | 21 ++++++++++-
 docs/execution-suspension.md               | 41 ++++++++++++++++++++++
 tests/test_loop.py                         |  2 +-
 5 files changed, 94 insertions(+), 5 deletions(-)
 create mode 100644 docs/execution-suspension.md

diff --git a/README.md b/README.md
index 045d354..a5426dc 100644
--- a/README.md
+++ b/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.
+
 `JobStore.recover(transcript)` compares each saved call and result with the
 checkpoint. A returned job whose exact evidence is already present reconciles
 silently; the absence of an old display notice is not a recovery event. Missing
diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py
index b2bdbef..5a236e5 100644
--- a/amplifier_module_loop_live/orchestrator.py
+++ b/amplifier_module_loop_live/orchestrator.py
@@ -201,6 +201,7 @@ async def _execute_live(self, prompt, context, providers, tools, hooks, coordina
         native = self._select_provider(providers)
         self.host.prepare_provider(native, coordinator)
         native = native if getattr(native, "native_bundle_live", False) else None
+        runtime.execution_suspension_support = native is None
         runtime.anchored_steering_mode = "unavailable" if native else "request_boundary"
         runtime.steering_wake = lambda: self.steer(_WAKE)
         coordinator.register_capability("live.steering", {"version": 1,
@@ -278,6 +279,7 @@ def restored_input(row):
             runtime.pause_epoch = restored["epoch"]
             runtime.pause_origin = restored.get("pause_origin")
             runtime.pause_terminal = restored.get("pause_terminal")
+            runtime.pause_jobs = restored.get("pause_jobs", {})
             runtime.generation = restored.get("generation")
             runtime.suspended = restored.get("turn")
             self.pending.extend(restored_input(row) for row in restored.get("pending", []))
@@ -285,6 +287,9 @@ def restored_input(row):
             runtime.pending_steering.extend(restored_input(row) for row in restored.get("steering", []))
             runtime.steering_outcomes.update(restored.get("steering_outcomes", {}))
             runtime.steering_closed = restored.get("steering_closed", False)
+            runtime.owned_jobs = {row["job_id"]: dict(row) for row in
+                [*restored.get("jobs", []), *runtime.pause_jobs.values()]}
+        self.pending_job_results = dict((restored or {}).get("pending_job_results", {}))
 
         async def paused_checkpoint():
             save = coordinator.get_capability("live.suspend")
@@ -294,6 +299,9 @@ def command_data(command):
                 return {key: value for key, value in vars(command).items() if key != "activation"}
             envelope = {"version": 1, "epoch": runtime.pause_epoch,
                         "pause_origin": runtime.pause_origin, "pause_terminal": runtime.pause_terminal,
+                        "pause_jobs": runtime.pause_jobs,
+                        "pending_job_results": {command.id: self.pending_job_results[command.id]
+                            for command in self.pending if command.id in self.pending_job_results},
                         "generation": runtime.generation, "turn": runtime.suspended,
                         "pending": [command_data(row) for row in self.pending],
                         "accepted": [command_data(row) for row in runtime.accepted.values()],
@@ -324,10 +332,12 @@ def command_data(command):
                     await self._hold_anchored("generation-ended")
                 if active is None and self.pending and runtime.inbox.empty() and not runtime.pause_epoch:
                     command = self.pending.popleft()
+                    origin = self.pending_job_results.pop(command.id, None)
                     # Publish busy state before scheduling: a host control can
                     # otherwise park after dequeue but before turn() first runs.
                     await runtime.emit("generation.started", generation_id=str(uuid.uuid4()),
-                                       initial_input_id=command.id, call_id=command.call_id)
+                                       initial_input_id=command.id, call_id=command.call_id,
+                                       **({"result_origin": origin} if origin else {}))
                     active = asyncio.create_task(turn(command))
                     idle = False
                 if active is None and not self.pending and not self._active_jobs() and runtime.inbox.empty() and not idle and not runtime.pause_epoch:
@@ -379,6 +389,16 @@ def command_data(command):
                     # turn keeps its generation and original input identity.
                     saved = runtime.suspended
                     runtime.pause_epoch, runtime.suspended = None, None
+                    # Durable inputs never contain a process-local writer
+                    # token. Only the explicit, admitted continuation handoff
+                    # may bind their original identities to this acquisition.
+                    if activation:
+                        from dataclasses import replace
+                        self.pending = deque(replace(command, activation=activation.current())
+                                             for command in self.pending)
+                        runtime.pending_steering = deque(
+                            replace(command, activation=activation.current())
+                            for command in runtime.pending_steering)
                     if saved:
                         command = restored_input(saved["command"])
                         if activation:
@@ -406,6 +426,8 @@ def command_data(command):
                     idle = False
                     continue
                 if kind == "bundle_job_suspended":
+                    if value in runtime.pause_jobs:
+                        runtime.pause_jobs[value]["status"] = "paused"
                     idle = False
                     continue
                 if kind == "input":
@@ -659,9 +681,13 @@ async def _job_returned(self, job_id, result, outcome, injections=(), *, deliver
         if deliver:
             self._pending_ephemeral_injections.extend(injections)
             activation = self.coordinator.get_capability("live.activation")
-            self.pending.append(Input("service", json.dumps({"job_id": job_id,
+            command = Input("service", json.dumps({"job_id": job_id,
                 "call_id": job["call_id"], "status": outcome, "tool_report": result}), source="amplifier-delegate",
-                activation=activation.current() if activation else None))
+                activation=activation.current() if activation else None)
+            self.pending.append(command)
+            self.pending_job_results[command.id] = {"jobId": job_id, "callId": job["call_id"]}
+            await self.runtime.emit("job.result_queued", input_id=command.id, job_id=job_id,
+                                    call_id=job["call_id"])
             self.steer(_WAKE)
 
 
diff --git a/amplifier_module_loop_live/runtime.py b/amplifier_module_loop_live/runtime.py
index f3cd5cf..3035a8a 100644
--- a/amplifier_module_loop_live/runtime.py
+++ b/amplifier_module_loop_live/runtime.py
@@ -73,16 +73,22 @@ def __init__(self, session_id=None, observer=None, max_input_chars=16000):
         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)
@@ -120,7 +126,7 @@ 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"}:
+        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:
@@ -178,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")
diff --git a/docs/execution-suspension.md b/docs/execution-suspension.md
new file mode 100644
index 0000000..dfc6d27
--- /dev/null
+++ b/docs/execution-suspension.md
@@ -0,0 +1,41 @@
+# 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.
+
+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.
diff --git a/tests/test_loop.py b/tests/test_loop.py
index c73e949..b5cf389 100644
--- a/tests/test_loop.py
+++ b/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={}),

From 9749172737176df18c35e22045ebd937a2394cb1 Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Wed, 21 Oct 2026 22:13:56 -0700
Subject: [PATCH 3/7] test: expose Live suspension recovery precondition gaps

---
 tests/test_suspension_recovery_boundaries.py | 171 +++++++++++++++++++
 1 file changed, 171 insertions(+)
 create mode 100644 tests/test_suspension_recovery_boundaries.py

diff --git a/tests/test_suspension_recovery_boundaries.py b/tests/test_suspension_recovery_boundaries.py
new file mode 100644
index 0000000..7e556d0
--- /dev/null
+++ b/tests/test_suspension_recovery_boundaries.py
@@ -0,0 +1,171 @@
+"""Recovery preconditions at the real Live loop boundary, without model calls."""
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from amplifier_core import HookRegistry, HookResult, ToolResult
+from amplifier_core.message_models import ToolCall
+from amplifier_module_loop_live.orchestrator import BundleLiveOrchestrator
+from amplifier_module_loop_live.runtime import Input, Runtime
+from amplifier_module_loop_live.job_store import JobStore
+
+
+def setup(restored=None, ledger=None):
+    runtime = Runtime(session_id="root")
+    capabilities = {"live.runtime": runtime, "live.suspend": AsyncMock(return_value=True),
+                    "live.maintenance_admit": AsyncMock(return_value=True)}
+    if restored is not None:
+        capabilities["live.suspended_execution"] = restored
+    if ledger is not None:
+        capabilities["live.jobs"] = ledger
+    coordinator = SimpleNamespace(
+        session_id="root", config={}, session_state={}, hooks=HookRegistry(),
+        cancellation=SimpleNamespace(is_cancelled=False, register_tool_start=Mock(),
+                                     register_tool_complete=Mock()),
+        get_capability=capabilities.get,
+        register_capability=lambda name, value: capabilities.__setitem__(name, value),
+        process_hook_result=AsyncMock(return_value=HookResult()),
+    )
+    context = SimpleNamespace(get_messages=AsyncMock(return_value=[]), set_messages=AsyncMock())
+    provider = SimpleNamespace(name="fixture", get_info=lambda: SimpleNamespace(default_model="fixture"))
+    loop = BundleLiveOrchestrator({"background_tools": ["effect"]})
+    # Result-observation turns are not the subject of these recovery tests.
+    loop._execute_guarded_goal = AsyncMock(return_value="observed")
+    return loop, runtime, coordinator, context, provider, capabilities
+
+
+async def start_or_fail(parts, tools=None):
+    loop, runtime, coordinator, context, provider, _ = parts
+    task = asyncio.create_task(loop.execute("", context, {"fixture": provider}, tools or {},
+                                            coordinator.hooks, coordinator))
+    notice = asyncio.create_task(runtime.wait_for(lambda e: e["type"] == "session.suspended", 2))
+    try:
+        done, _ = await asyncio.wait((task, notice), timeout=3, return_when=asyncio.FIRST_COMPLETED)
+        assert done, "loop neither suspended nor refused"
+        if task in done:
+            await task
+        await notice
+        return task
+    except BaseException:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+        raise
+    finally:
+        notice.cancel()
+        await asyncio.gather(notice, return_exceptions=True)
+
+
+async def finish(task, runtime):
+    if not task.done():
+        await runtime.submit(Input("stop"))
+    await asyncio.wait_for(task, 2)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("version", [None, 2, True])
+async def test_unsupported_envelope_refuses_before_restore_or_ack(version):
+    envelope = {"version": version, "epoch": "held", "pending": []}
+    parts = setup(envelope)
+    task = None
+    try:
+        with pytest.raises(ValueError, match="version"):
+            task = await start_or_fail(parts)
+        assert parts[1].pause_epoch is None
+        parts[5]["live.suspend"].assert_not_awaited()
+    finally:
+        if task:
+            await finish(task, parts[1])
+
+
+@pytest.mark.asyncio
+async def test_no_ledger_cannot_acknowledge_paused_background_job():
+    parts = setup()
+    loop, runtime, _, _, _, capabilities = parts
+    done = asyncio.get_running_loop().create_future()
+    done.set_result(None)
+    loop.jobs["job"] = {"task": done, "call_id": "call", "status": "paused",
+                        "continuation": {"version": 1, "child": "child"}}
+    await runtime.request_pause("held")
+    task = None
+    try:
+        with pytest.raises(RuntimeError, match="ledger"):
+            task = await start_or_fail(parts)
+        capabilities["live.suspend"].assert_not_awaited()
+    finally:
+        if task:
+            await finish(task, runtime)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("change", ["missing", "continuation", "extra"])
+async def test_restored_jobs_must_match_durable_ledger(tmp_path, change):
+    store = JobStore(tmp_path / "jobs")
+    call = ToolCall(id="call", name="effect", arguments={})
+    continuation = {"version": 1, "child": "child"}
+    store.begin(call, "job", "receipt")
+    store.suspend(call.id, continuation)
+    row = {"job_id": "job", "call_id": "call", "status": "paused", "continuation": continuation}
+    envelope = {"version": 1, "epoch": "held", "jobs": [row]}
+    if change == "missing":
+        ledger = None
+    else:
+        ledger = store
+        if change == "continuation":
+            row["continuation"] = {"version": 1, "child": "changed"}
+        else:
+            envelope["jobs"] = []
+    parts = setup(envelope, ledger)
+    task = None
+    try:
+        with pytest.raises(RuntimeError, match="ledger"):
+            task = await start_or_fail(parts)
+        parts[5]["live.suspend"].assert_not_awaited()
+    finally:
+        if task:
+            await finish(task, parts[1])
+        store.close()
+
+
+@pytest.mark.asyncio
+async def test_two_paused_jobs_remount_once_with_original_group(tmp_path):
+    store = JobStore(tmp_path / "jobs")
+    calls = [ToolCall(id=f"call-{i}", name="effect", arguments={}) for i in range(2)]
+    envelope = {"version": 1, "epoch": "held", "jobs": []}
+    for i, call in enumerate(calls):
+        continuation = {"version": 1, "child": f"child-{i}"}
+        store.begin(call, f"job-{i}", "receipt", parallel_group_id="original-group")
+        store.suspend(call.id, continuation)
+        envelope["jobs"].append({"job_id": f"job-{i}", "call_id": call.id,
+                                  "status": "paused", "continuation": continuation})
+    store.close()
+    store = JobStore(tmp_path / "jobs")
+    parts = setup(envelope, store)
+    _, runtime, coordinator, _, _, _ = parts
+    resumed = []
+
+    class Effect:
+        name = "effect"
+        input_schema = {"type": "object", "properties": {}}
+        async def execute(self, arguments):
+            pytest.fail("suspended operation must not be redispatched")
+        async def resume_execution(self, continuation):
+            dispatch = coordinator._tool_dispatch_contexts[asyncio.current_task()]
+            resumed.append((continuation["child"], dict(dispatch)))
+            return ToolResult(success=True, output=continuation["child"])
+
+    task = await start_or_fail(parts, {"effect": Effect()})
+    try:
+        assert resumed == []
+        await runtime.resume_paused("held")
+        await runtime.wait_for(lambda e: e["type"] == "job.returned" and e["job_id"] == "job-1", 2)
+        assert sorted(child for child, _ in resumed) == ["child-0", "child-1"]
+        assert {d["tool_call_id"] for _, d in resumed} == {"call-0", "call-1"}
+        assert {d["parallel_group_id"] for _, d in resumed} == {"original-group"}
+        assert all(row["status"] == "returned" for row in store.rows.values())
+        with pytest.raises(RuntimeError, match="changed"):
+            await runtime.resume_paused("held")
+        assert len(resumed) == 2
+    finally:
+        await finish(task, runtime)
+        store.close()

From a301964f0d2720f2c1063a8d17a71dd0cf082dd1 Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Wed, 21 Oct 2026 22:15:06 -0700
Subject: [PATCH 4/7] fix: validate durable Live suspension ownership before
 remount

---
 amplifier_module_loop_live/job_store.py    |  3 +-
 amplifier_module_loop_live/orchestrator.py | 41 +++++++++++++++++++---
 docs/execution-suspension.md               | 14 ++++++++
 tests/test_continuation_guard.py           |  2 +-
 4 files changed, 54 insertions(+), 6 deletions(-)

diff --git a/amplifier_module_loop_live/job_store.py b/amplifier_module_loop_live/job_store.py
index 5956ef3..1a36e0c 100644
--- a/amplifier_module_loop_live/job_store.py
+++ b/amplifier_module_loop_live/job_store.py
@@ -68,11 +68,12 @@ def _write(self, row):
                 os.unlink(temporary)
         self.rows[row["call_id"]] = row
 
-    def begin(self, call, job_id, receipt, native_call=None, *, generation=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",
+                     "parallel_group_id": parallel_group_id,
                      "generation": copy.deepcopy(generation)})
 
     def finish(self, call_id, result, outcome):
diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py
index 5a236e5..bfe703f 100644
--- a/amplifier_module_loop_live/orchestrator.py
+++ b/amplifier_module_loop_live/orchestrator.py
@@ -38,6 +38,34 @@ def __init__(self, config=None):
         self.root_provider = None
         self.approved_dispatch = ApprovedDispatch(self)
 
+    @staticmethod
+    def _validate_suspended_envelope(envelope, ledger):
+        """Validate portable format and durable job ownership before restore."""
+        if not isinstance(envelope, dict) or type(envelope.get("version")) is not int or envelope["version"] != 1:
+            raise ValueError("Unsupported live suspension envelope version")
+        epoch = envelope.get("epoch")
+        if not isinstance(epoch, str) or not 1 <= len(epoch) <= 128:
+            raise ValueError("Invalid live suspension epoch")
+        jobs = envelope.get("jobs", [])
+        if not isinstance(jobs, list) or any(not isinstance(row, dict) for row in jobs):
+            raise ValueError("Invalid live suspension jobs")
+        paused = {row["job_id"]: row for row in (ledger.rows.values() if ledger else ())
+                  if row.get("status") == "paused"}
+        if jobs and ledger is None:
+            raise RuntimeError("Paused background jobs require a durable live.jobs ledger")
+        seen = set()
+        for job in jobs:
+            identity = job.get("job_id")
+            row = paused.get(identity) if isinstance(identity, str) else None
+            if (row is None or identity in seen or job.get("status") != "paused"
+                    or job.get("call_id") != row.get("call_id")
+                    or job.get("continuation") is None
+                    or job["continuation"] != row.get("continuation")):
+                raise RuntimeError("Suspended job does not match its durable ledger")
+            seen.add(identity)
+        if seen != set(paused):
+            raise RuntimeError("Suspension envelope omits a paused ledger job")
+
     async def _execute_guarded_goal(self, prompt, context, providers, tools, hooks, coordinator, resume=None):
         # An explicit input always receives its first turn. Only automatic goal
         # continuations consult the host at the last safe admission boundary.
@@ -177,6 +205,10 @@ async def _execute_live(self, prompt, context, providers, tools, hooks, coordina
                 raise
         if self.running:
             raise RuntimeError("One execute owner per manager session")
+        ledger = coordinator.get_capability("live.jobs")
+        restored = coordinator.get_capability("live.suspended_execution")
+        if restored is not None:
+            self._validate_suspended_envelope(restored, ledger)
         self.running, self.runtime = True, runtime
         self.context, self.tools, self.hooks, self.coordinator = context, tools, hooks, coordinator
         from .jobs import AsyncTool, JobControl
@@ -189,7 +221,6 @@ async def _execute_live(self, prompt, context, providers, tools, hooks, coordina
             {"job_id": identity, "call_id": job["call_id"], "status": job.get("status", "pending"),
              "tool": (job.get("tool_call") or {}).get("name")}
             for identity, job in self.jobs.items() if job.get("result") is None])
-        ledger = coordinator.get_capability("live.jobs")
         if ledger:
             recovered = coordinator.get_capability("live.recovered_jobs") or []
             for row in ledger.rows.values():
@@ -272,7 +303,6 @@ async def turn(command, *, continuation=False, resume=None):
                     activation.reset(activation_token)
                 LIVE_OWNER.reset(owner_token)
 
-        restored = coordinator.get_capability("live.suspended_execution")
         def restored_input(row):
             return Input(**{**row, "attachments": tuple(row.get("attachments", ()))})
         if restored:
@@ -312,6 +342,7 @@ def command_data(command):
                                   "status": job.get("status"), "continuation": job.get("continuation"),
                                   "generation": job.get("generation")}
                                  for identity, job in self.jobs.items() if job.get("result") is None]}
+            self._validate_suspended_envelope(envelope, ledger)
             if await save(envelope) is not True:
                 raise RuntimeError("Live suspension checkpoint was not acknowledged")
             return envelope
@@ -413,7 +444,8 @@ def command_data(command):
                                 ledger.resume_paused(call.id, continuation)
                             job["status"] = "pending"
                             job["task"] = asyncio.create_task(self._run_background_job(
-                                job_id, call, tools, hooks, None, coordinator, continuation=continuation))
+                                job_id, call, tools, hooks, job.get("parallel_group_id"),
+                                coordinator, continuation=continuation))
                     idle = False
                     await runtime.emit("session.resumed", epoch=value)
                     continue
@@ -613,11 +645,12 @@ async def _execute_tool_scoped(self, tool_call, tools, hooks, parallel_group_id,
             provider = NATIVE_REQUEST.get()
             ledger.begin(tool_call, job_id, receipt,
                          provider.async_calls.get(tool_call.id) if provider else None,
-                         generation=runtime.generation)
+                         generation=runtime.generation, parallel_group_id=parallel_group_id)
 
 
         self.jobs[job_id] = {"task": asyncio.create_task(self._run_background_job(job_id, tool_call, tools, hooks, parallel_group_id, coordinator)), "call_id": tool_call.id,
                              "tool_call": tool_call.model_dump(), "receipt": receipt, "status": "pending",
+                             "parallel_group_id": parallel_group_id,
                              "generation": __import__("copy").deepcopy(runtime.generation)}
         self._tool_calls_this_turn += 1
         origin = self.jobs[job_id]["generation"] or {}
diff --git a/docs/execution-suspension.md b/docs/execution-suspension.md
index dfc6d27..24371f5 100644
--- a/docs/execution-suspension.md
+++ b/docs/execution-suspension.md
@@ -35,6 +35,20 @@ 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 before restoring runtime state.
+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
diff --git a/tests/test_continuation_guard.py b/tests/test_continuation_guard.py
index d711bfb..67420d4 100644
--- a/tests/test_continuation_guard.py
+++ b/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')])

From b80d0093c5d9a1909e6f3739f2397a4a5bffe50b Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Wed, 21 Oct 2026 22:22:00 -0700
Subject: [PATCH 5/7] test: cover missing envelope and post-refusal retained
 state

---
 tests/test_suspension_recovery_boundaries.py | 55 ++++++++++++++++++--
 1 file changed, 52 insertions(+), 3 deletions(-)

diff --git a/tests/test_suspension_recovery_boundaries.py b/tests/test_suspension_recovery_boundaries.py
index 7e556d0..9c26036 100644
--- a/tests/test_suspension_recovery_boundaries.py
+++ b/tests/test_suspension_recovery_boundaries.py
@@ -39,7 +39,8 @@ async def start_or_fail(parts, tools=None):
     loop, runtime, coordinator, context, provider, _ = parts
     task = asyncio.create_task(loop.execute("", context, {"fixture": provider}, tools or {},
                                             coordinator.hooks, coordinator))
-    notice = asyncio.create_task(runtime.wait_for(lambda e: e["type"] == "session.suspended", 2))
+    notice = asyncio.create_task(runtime.wait_for(
+        lambda e: e["type"] in {"session.suspended", "session.idle"}, 2))
     try:
         done, _ = await asyncio.wait((task, notice), timeout=3, return_when=asyncio.FIRST_COMPLETED)
         assert done, "loop neither suspended nor refused"
@@ -92,13 +93,16 @@ async def test_no_ledger_cannot_acknowledge_paused_background_job():
         with pytest.raises(RuntimeError, match="ledger"):
             task = await start_or_fail(parts)
         capabilities["live.suspend"].assert_not_awaited()
+        assert runtime.closed and runtime.events[-1]["status"] == "error"
+        assert loop.jobs["job"]["continuation"] == {"version": 1, "child": "child"}
+        assert not any(e["type"] == "session.suspended" for e in runtime.events)
     finally:
         if task:
             await finish(task, runtime)
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("change", ["missing", "continuation", "extra"])
+@pytest.mark.parametrize("change", ["missing", "continuation", "extra", "no-envelope", "generation"])
 async def test_restored_jobs_must_match_durable_ledger(tmp_path, change):
     store = JobStore(tmp_path / "jobs")
     call = ToolCall(id="call", name="effect", arguments={})
@@ -113,14 +117,20 @@ async def test_restored_jobs_must_match_durable_ledger(tmp_path, change):
         ledger = store
         if change == "continuation":
             row["continuation"] = {"version": 1, "child": "changed"}
-        else:
+        elif change == "extra":
             envelope["jobs"] = []
+        elif change == "generation":
+            row["generation"] = {"id": "different-generation"}
+        else:
+            envelope = None
     parts = setup(envelope, ledger)
     task = None
     try:
         with pytest.raises(RuntimeError, match="ledger"):
             task = await start_or_fail(parts)
         parts[5]["live.suspend"].assert_not_awaited()
+        assert store.rows[call.id]["status"] == "paused"
+        assert store.rows[call.id]["continuation"] == continuation
     finally:
         if task:
             await finish(task, parts[1])
@@ -169,3 +179,42 @@ async def resume_execution(self, continuation):
     finally:
         await finish(task, runtime)
         store.close()
+
+
+@pytest.mark.parametrize("epoch", [None, 7, "", "a" * 129])
+def test_invalid_epoch_refuses(epoch):
+    with pytest.raises(ValueError, match="epoch"):
+        BundleLiveOrchestrator._validate_suspended_envelope(
+            {"version": 1, "epoch": epoch}, None)
+
+
+@pytest.mark.parametrize("jobs", [None, {}, ["not-a-row"]])
+def test_invalid_job_collection_refuses(jobs):
+    with pytest.raises(ValueError, match="jobs"):
+        BundleLiveOrchestrator._validate_suspended_envelope(
+            {"version": 1, "epoch": "held", "jobs": jobs}, None)
+
+
+@pytest.mark.parametrize("change", ["duplicate", "call", "status", "null-continuation"])
+def test_invalid_job_reference_refuses(tmp_path, change):
+    store = JobStore(tmp_path)
+    call = ToolCall(id="call", name="effect", arguments={})
+    store.begin(call, "job", "receipt")
+    store.suspend(call.id, {"child": "held"})
+    row = {"job_id": "job", "call_id": "call", "status": "paused",
+           "continuation": {"child": "held"}}
+    envelope = {"version": 1, "epoch": "held", "jobs": [row]}
+    if change == "duplicate":
+        envelope["jobs"].append(dict(row))
+    elif change == "call":
+        row["call_id"] = "other"
+    elif change == "status":
+        row["status"] = "pending"
+    else:
+        row["continuation"] = None
+    try:
+        with pytest.raises(RuntimeError, match="ledger"):
+            BundleLiveOrchestrator._validate_suspended_envelope(envelope, store)
+        assert store.rows[call.id]["status"] == "paused"
+    finally:
+        store.close()

From aee0f4371a01727fe508908225b98fff67ff041d Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Wed, 21 Oct 2026 22:22:47 -0700
Subject: [PATCH 6/7] fix: refuse orphaned paused jobs before Live activation

---
 amplifier_module_loop_live/orchestrator.py | 12 ++++++++++--
 docs/execution-suspension.md               |  4 +++-
 2 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py
index bfe703f..5db8659 100644
--- a/amplifier_module_loop_live/orchestrator.py
+++ b/amplifier_module_loop_live/orchestrator.py
@@ -49,8 +49,13 @@ def _validate_suspended_envelope(envelope, ledger):
         jobs = envelope.get("jobs", [])
         if not isinstance(jobs, list) or any(not isinstance(row, dict) for row in jobs):
             raise ValueError("Invalid live suspension jobs")
-        paused = {row["job_id"]: row for row in (ledger.rows.values() if ledger else ())
-                  if row.get("status") == "paused"}
+        paused = {}
+        for row in (ledger.rows.values() if ledger is not None else ()):
+            if row.get("status") == "paused":
+                identity = row.get("job_id")
+                if not isinstance(identity, str) or not identity or identity in paused:
+                    raise RuntimeError("Invalid paused ledger job identity")
+                paused[identity] = row
         if jobs and ledger is None:
             raise RuntimeError("Paused background jobs require a durable live.jobs ledger")
         seen = set()
@@ -59,6 +64,7 @@ def _validate_suspended_envelope(envelope, ledger):
             row = paused.get(identity) if isinstance(identity, str) else None
             if (row is None or identity in seen or job.get("status") != "paused"
                     or job.get("call_id") != row.get("call_id")
+                    or job.get("generation") != row.get("generation")
                     or job.get("continuation") is None
                     or job["continuation"] != row.get("continuation")):
                 raise RuntimeError("Suspended job does not match its durable ledger")
@@ -209,6 +215,8 @@ async def _execute_live(self, prompt, context, providers, tools, hooks, coordina
         restored = coordinator.get_capability("live.suspended_execution")
         if restored is not None:
             self._validate_suspended_envelope(restored, ledger)
+        elif ledger is not None and any(row.get("status") == "paused" for row in ledger.rows.values()):
+            raise RuntimeError("Paused ledger jobs require their suspension envelope")
         self.running, self.runtime = True, runtime
         self.context, self.tools, self.hooks, self.coordinator = context, tools, hooks, coordinator
         from .jobs import AsyncTool, JobControl
diff --git a/docs/execution-suspension.md b/docs/execution-suspension.md
index 24371f5..964d6b5 100644
--- a/docs/execution-suspension.md
+++ b/docs/execution-suspension.md
@@ -38,7 +38,9 @@ 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 before restoring runtime state.
+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.

From d52f0ecf17e47e9e88f6d4d5c029e75156255ca4 Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Wed, 28 Oct 2026 16:42:03 -0700
Subject: [PATCH 7/7] build: resolve accepted Streaming suspension dependency

---
 uv.lock | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/uv.lock b/uv.lock
index 6316a38..d89cde9 100644
--- a/uv.lock
+++ b/uv.lock
@@ -61,7 +61,7 @@ dev = [
 [[package]]
 name = "amplifier-module-loop-streaming"
 version = "1.0.0"
-source = { git = "https://github.com/microsoft/amplifier-module-loop-streaming?rev=main#9e5d4ff07c7f57fbdf233b869d94e18c910d83b9" }
+source = { git = "https://github.com/microsoft/amplifier-module-loop-streaming?rev=main#941806a4a42c84c7c239fd8e0f8ef1260f1ec32f" }
 
 [[package]]
 name = "annotated-types"