Preserve accepted steering through generation completion

amplifier_module_loop_live/failures.py

@@ -25,7 +25,17 @@ def turn_failure(error, stage="manager_turn"):
     module = origin.tb_frame.f_globals.get("__name__", "") if origin else ""
     local_context = (module.startswith("amplifier_module_context_")
                      or module == "amplifier_module_loop_streaming")
-    if isinstance(error, ContextLengthError):
+    compaction = (type(error).__module__ == "amplifier_module_context_managed.errors"
+                  and type(error).__name__ == "CompactionError")
+    code = getattr(error, "code", None) if compaction else None
+    safe_codes = {"native_input_oversized", "native_checkpoint_invalid", "native_no_reduction",
+                  "native_measurement_unavailable", "native_compaction_failed",
+                  "disabled", "request_context_unavailable", "invalid_native_contract",
+                  "authoritative_measurement_unavailable"}
+    if compaction:
+        category, stage = "context_compaction", "context_preparation"
+        message = "Context compaction failed. Original history is preserved; repair context preparation before continuing."
+    elif isinstance(error, ContextLengthError):
         category = "context_limit"
         if local_context:
             stage = "context_preparation"
@@ -49,7 +59,8 @@ def turn_failure(error, stage="manager_turn"):
     # Class names from arbitrary third-party exceptions are not trusted text.
     kind = next((cls.__name__ for cls in type(error).__mro__
                  if cls.__module__ in {"builtins", "amplifier_core.llm_errors"}), "Exception")
-    return {"error_type": kind, "error_category": category, "error_stage": stage,
+    return {**({"error_code": code} if code in safe_codes else {}),
+            "error_type": "CompactionError" if compaction else kind, "error_category": category, "error_stage": stage,
             "error_message": message, "effects": "not_rolled_back", "replayed": False,
             "retryable": isinstance(error, LLMError) and error.retryable is True}
 

amplifier_module_loop_live/jobs.py

@@ -79,7 +79,8 @@ async def execute(self, input):
                 runtime = self.loop.runtime
                 cursor = runtime.sequence
                 # Input already queued before the wait must also release it.
-                if any(command.kind in {"user", "steer"} for command in self.loop.pending) or runtime.queued_inputs:
+                if (any(command.kind in {"user", "steer"} for command in self.loop.pending)
+                        or runtime.pending_steering or runtime.queued_inputs):
                     reason = "input_pending"
                 else:
                     try:

amplifier_module_loop_live/orchestrator.py

@@ -15,7 +15,7 @@
 from amplifier_core import HookResult
 from amplifier_module_loop_streaming import StreamingOrchestrator, ConversationProviderPin
 from .runtime import Input, observation
-from .provenance import InputContext, input_metadata
+from .provenance import InputContext, ContinuationContext, input_metadata
 from .job_store import serialize_result
 from .host import HostAdapter, AttachmentContext
 from .scope import HOST_ADAPTER
@@ -66,6 +66,8 @@ def _select_provider(self, providers):
         return self.root_provider or super()._select_provider(providers)
 
     async def _hold_anchored(self, reason):
+        self.pending.extend(self.runtime.pending_steering)
+        self.runtime.pending_steering.clear()
         retained = deque()
         for command in self.pending:
             if command.target_generation_id is None:
@@ -82,6 +84,8 @@ async def _drain_steering(self, context, hooks, iteration):
         # The base queue is a wake signal; our queue retains identities across
         # the base engine's turn-start and cancellation queue resets.
         legacy = [text for text in self._steering_queue.drain() if text != _WAKE]
+        self.pending.extend(self.runtime.pending_steering)
+        self.runtime.pending_steering.clear()
         commands = list(self.pending)
         self.pending.clear()
         applied = []
@@ -180,6 +184,7 @@ async def _execute_live(self, prompt, context, providers, tools, hooks, coordina
         self.host.prepare_provider(native, coordinator)
         native = native if getattr(native, "native_bundle_live", False) else 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,
             "mode": runtime.anchored_steering_mode, "cancellable": False,
             "submit": runtime.submit_steering})
@@ -213,7 +218,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):
+        async def turn(command, *, continuation=False):
             activation = coordinator.get_capability("live.activation")
             activation_token = None
             owner_token = LIVE_OWNER.set(self)
@@ -222,15 +227,18 @@ async def turn(command):
                 if activation:
                     activation_token = activation.bind(command.activation)
                 await self._synchronize_job_results(context)
-                await runtime.emit("input.delivered", input_id=command.id,
-                                   delivery="new_turn", source=command.source)
                 turn_context = context
-                if command.attachments:
-                    turn_context = AttachmentContext(context, self._text(command), self.host.content(command, coordinator))
-                turn_context = InputContext(turn_context, self._text(command), command)
+                if continuation:
+                    turn_context = ContinuationContext(context)
+                else:
+                    await runtime.emit("input.delivered", input_id=command.id,
+                                       delivery="new_turn", source=command.source)
+                    if command.attachments:
+                        turn_context = AttachmentContext(context, self._text(command), self.host.content(command, coordinator))
+                    turn_context = InputContext(turn_context, self._text(command), command)
                 stage = "manager_turn"
                 result = await self._execute_guarded_goal(
-                    self._text(command), turn_context, providers, tools, hooks, coordinator)
+                    "" if continuation else self._text(command), turn_context, providers, tools, hooks, coordinator)
                 await runtime.inbox.put(("bundle_turn", (result, None)))
             except asyncio.CancelledError:
                 raise
@@ -319,10 +327,11 @@ async def turn(command):
                         await runtime.emit("input.queued", input_id=value.id, delivery="request_boundary")
                 elif kind == "bundle_turn":
                     await active
-                    active = None
                     result, error = value
-                    await self._hold_anchored("generation-failed" if error else "generation-completed")
                     if error:
+                        runtime.steering_closed = True
+                        active = None
+                        await self._hold_anchored("generation-failed")
                         cause, failure = error
                         await runtime.emit("generation.failed", **failure)
                         # The base engine owns provider-specific events. A local
@@ -332,6 +341,20 @@ async def turn(command):
                     checkpoint = coordinator.get_capability("live.checkpoint")
                     if checkpoint:
                         await checkpoint()
+                    # Accepted input can arrive during the final response or
+                    # its checkpoint. Keep the same generation alive for it;
+                    # 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
+                            and not coordinator.cancellation.is_cancelled and not self._budget_exhausted):
+                        active = asyncio.create_task(turn(pending[0], continuation=True))
+                        continue
+                    # No await between the empty-queue decision and closing
+                    # admission. Later submissions fail before being accepted.
+                    runtime.steering_closed = True
+                    active = None
+                    await self._hold_anchored("generation-completed")
                     messages = await context.get_messages()
                     final_message = messages[-1] if messages else {}
                     final_content = final_message.get("content", "") if final_message.get("role") == "assistant" and not final_message.get("tool_calls") else ""
@@ -362,6 +385,7 @@ async def turn(command):
             raise
         finally:
             runtime.closed = True
+            runtime.steering_wake = None
             if active:
                 active.cancel()
                 await asyncio.gather(active, return_exceptions=True)

amplifier_module_loop_live/provenance.py

@@ -21,3 +21,22 @@ async def add_message(self, message):
             message = {**message, 'metadata': {**(message.get('metadata') or {}), **input_metadata(self.command)}}
             self.pending = False
         return await self.context.add_message(message)
+
+
+class ContinuationContext:
+    """Resume an accepted turn without inventing or replaying a user prompt.
+
+    The base engine always appends its initial prompt. Real pending inputs are
+    appended by the normal steering drain, with their identities, on iteration 1.
+    """
+    def __init__(self, context):
+        self.context, self.pending = context, True
+
+    def __getattr__(self, name):
+        return getattr(self.context, name)
+
+    async def add_message(self, message):
+        if self.pending and message.get('role') == 'user' and message.get('content') == '':
+            self.pending = False
+            return
+        return await self.context.add_message(message)

amplifier_module_loop_live/runtime.py

@@ -5,6 +5,7 @@
 import json
 import time
 import uuid
+from collections import deque
 from dataclasses import dataclass, field, replace
 
 
@@ -64,6 +65,10 @@ def __init__(self, session_id=None, observer=None, max_input_chars=16000):
         self.generation = None
         self.anchored_steering_mode = None
         self.steering_outcomes = {}
+        self.pending_steering = deque()
+        self.steering_wake = None
+        self.steering_closed = False
+        self.stop_requested = False
 
     async def submit(self, command: Input):
         if self.closed:
@@ -94,18 +99,33 @@ async def submit(self, command: Input):
                 raise ValueError("Anchored steering is unavailable for this execution mode")
             if not self.generation or self.generation["id"] != command.target_generation_id:
                 raise ValueError("The anchored generation is no longer active")
+            if self.steering_closed:
+                raise ValueError("The anchored generation is finishing and no longer accepts steering")
         # Never silently evict identities and make an old command executable again.
         if len(self.accepted) >= 2000:
             raise RuntimeError("Session input limit reached; start a new session")
-        if self.queued_inputs >= 128:
+        if self.queued_inputs + len(self.pending_steering) >= 128:
             raise asyncio.QueueFull()
-        self.inbox.put_nowait(("input", command))
-        self.queued_inputs += 1
+        direct = command.target_generation_id is not None and self.steering_wake is not None
+        if direct:
+            # Admission and the request-boundary queue share one synchronous
+            # step. A busy manager inbox cannot postpone an accepted steer.
+            self.pending_steering.append(command)
+            self.steering_wake()
+        else:
+            self.inbox.put_nowait(("input", command))
+            self.queued_inputs += 1
         self.accepted[command.id] = command
+        if command.kind == "stop":
+            self.stop_requested = True
         if command.target_generation_id is not None:
             self.steering_outcomes[command.id] = {"accepted": True, "inputId": command.id, "generationId": command.target_generation_id, "disposition": "queued"}
         await self.emit("input.accepted", input_id=command.id, kind=command.kind,
                         source=command.source, target=command.target)
+        if direct:
+            await self.emit("steering.accepted", input_id=command.id,
+                            target_generation_id=command.target_generation_id, delivery="request_boundary")
+            await self.emit("input.queued", input_id=command.id, delivery="request_boundary")
         return command.id
 
     async def submit_steering(self, command: Input, generation_id: str):
@@ -125,6 +145,7 @@ async def emit(self, event_type, **data):
         # Optional portable generation correlation. A generation is one finite
         # manager turn; it may finish while delegated jobs are still running.
         if event_type == "generation.started":
+            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 generation is not None:

docs/anchored-steering.md

@@ -31,8 +31,18 @@ with `input_id`, `target_generation_id`, and `delivery: "request_boundary"`.
 a reason. Context input metadata also retains `target_generation_id`.
 Generation terminal events keep the existing distinction between delivered and
 accepted-but-undelivered input IDs. A stale or unavailable generation is rejected
-before admission. Once admitted, a finishing generation holds late input; it
-never converts that input into a fresh turn.
+before admission. Admission places anchored input directly in the boundary
+queue and wakes the engine, without waiting for the manager to consume an inbox
+event. Each boundary drains the pending batch in order.
+
+A normally finishing generation checks again after its durable checkpoint. If
+accepted steering remains, it continues the same generation with those inputs,
+without replaying the original prompt or completed tools. Otherwise it closes
+admission synchronously before terminal publication: racing submissions are
+rejected before acceptance. Stop, cancellation, failure, and exhausted call
+budgets hold pending input instead of resuming work. No input is silently moved
+to a new generation. An acknowledgment proves context insertion, not immediate
+model attention; a synchronous tool must still return before the next request.
 
 Provider-native `steer_live` is deliberately unavailable through this contract
 until it can supply a matching attribution and withdrawal/effect protocol.

docs/generation-events.md

@@ -14,6 +14,8 @@ The optional live runtime now publishes a bounded manager-generation contract:
 
 Finished `text` is the final ordinary assistant message from the Amplifier context, without tool calls. It excludes provisional messages that the base streaming engine may concatenate into its returned string. Empty text is valid and does not mean work completed successfully.
 
+A generation stays open through its final checkpoint while accepted request-boundary steering remains. Such input can require another engine invocation within the same generation, without another `generation.started` or replay of earlier work. Terminal publication closes admission before awaiting context reads.
+
 A finished manager generation is **not** completion of all delegated work. `active_job_ids` identifies still-running local delegate jobs. Their results reenter the manager through service observations. A host should show the manager's answer plus pending work, and continue processing later generations. Do not replay a job to obtain its result.
 
 This contract changes no finite-session behavior, imports no app/CLI/provider SDK, and adds no provider protocol fields. Existing consumers can ignore the added metadata. Fixtures cover steering during a request, provisional text before a background tool, final text selection, native acceptance versus application, correlated failure, and cancellation without a success event.

tests/test_failures.py

@@ -57,3 +57,19 @@ async def test_local_failure_keeps_cause_and_identity_without_provider_error_or_
     assert 'generation.finished' not in [e['type'] for e in runtime.events]
     assert 'secret-key' not in json.dumps(runtime.events)
     loop._execute_guarded_goal.assert_awaited_once()
+
+
+@pytest.mark.parametrize('code', ['native_input_oversized', 'private payload'])
+def test_compaction_failure_preserves_bounded_code_without_optional_dependency(code):
+    error_class = type('CompactionError', (RuntimeError,), {
+        '__module__': 'amplifier_module_context_managed.errors'})
+    error = error_class('secret transcript and SDK response')
+    error.code = code
+    failure = turn_failure(error)
+    assert failure['error_category'] == 'context_compaction'
+    assert failure['error_stage'] == 'context_preparation'
+    assert failure['error_type'] == 'CompactionError'
+    assert failure.get('error_code') == ('native_input_oversized' if code != 'private payload' else None)
+    assert 'secret' not in json.dumps(failure)
+    assert 'private payload' not in json.dumps(failure)
+    assert failure['retryable'] is False