From 46723db070e31c68d847b8997285d88f2f06e669 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Tue, 6 Oct 2026 12:56:03 -0700 Subject: [PATCH 1/2] Drain accepted steering through the generation completion boundary --- amplifier_module_loop_live/jobs.py | 3 +- amplifier_module_loop_live/orchestrator.py | 44 +++++-- amplifier_module_loop_live/provenance.py | 19 +++ amplifier_module_loop_live/runtime.py | 27 ++++- docs/anchored-steering.md | 14 ++- docs/generation-events.md | 2 + tests/test_loop.py | 127 ++++++++++++++++++--- 7 files changed, 206 insertions(+), 30 deletions(-) diff --git a/amplifier_module_loop_live/jobs.py b/amplifier_module_loop_live/jobs.py index 16c35ec..9301287 100644 --- a/amplifier_module_loop_live/jobs.py +++ b/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: diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py index c14aae5..9a5bc17 100644 --- a/amplifier_module_loop_live/orchestrator.py +++ b/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) diff --git a/amplifier_module_loop_live/provenance.py b/amplifier_module_loop_live/provenance.py index 5301451..5c2c345 100644 --- a/amplifier_module_loop_live/provenance.py +++ b/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) diff --git a/amplifier_module_loop_live/runtime.py b/amplifier_module_loop_live/runtime.py index 40ab478..becaf76 100644 --- a/amplifier_module_loop_live/runtime.py +++ b/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: diff --git a/docs/anchored-steering.md b/docs/anchored-steering.md index 7a3d910..7d55912 100644 --- a/docs/anchored-steering.md +++ b/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. diff --git a/docs/generation-events.md b/docs/generation-events.md index 803aebc..fbc3b5d 100644 --- a/docs/generation-events.md +++ b/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. diff --git a/tests/test_loop.py b/tests/test_loop.py index 4672a18..c73e949 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -74,34 +74,115 @@ async def test_anchored_steering_applies_only_to_its_current_generation(self): finished = await self.runtime.wait_for(lambda e: e["type"] == "generation.finished" and e["generation_id"] == generation, 3) self.assertEqual(finished["input_ids"], ["original", "anchored"]) - async def test_anchored_steering_is_held_instead_of_starting_another_turn(self): + async def test_accepted_checkpoint_steering_continues_the_same_generation(self): await self.start() await self.runtime.submit(Input("user", "ORIGINAL", id="original")) await self.provider.request() generation = self.runtime.generation["id"] - command = Input("steer", "MUST NOT BECOME NEW WORK", id="held") checkpoint_entered, release_checkpoint = asyncio.Event(), asyncio.Event() async def checkpoint(*_args): checkpoint_entered.set() await release_checkpoint.wait() self.session.coordinator.register_capability("live.checkpoint", checkpoint) - await self.provider.reply("finite final response") + await self.provider.reply("first answer") await asyncio.wait_for(checkpoint_entered.wait(), 3) - await self.runtime.submit_steering(command, generation) + for i in range(3): + outcome = await self.runtime.submit_steering(Input("steer", f"LATE-{i}", id=f"late-{i}"), generation) + self.assertEqual(outcome["disposition"], "queued") release_checkpoint.set() - held = await self.runtime.wait_for(lambda e: e["type"] == "steering.held" and e["input_id"] == "held", 3) - self.assertEqual(held["reason"], "generation-ended") - await self.runtime.wait_for(lambda e: e["type"] == "session.idle", 3) - self.assertTrue(self.provider.requests.empty()) - self.assertNotIn(command.text, str(await self.session.coordinator.get("context").get_messages())) - self.assertEqual((await self.runtime.submit_steering(command, generation))["disposition"], "held") + request = await self.provider.request() + text = str(request.messages) + self.assertEqual(text.count("ORIGINAL"), 1) + positions = [text.index(f"LATE-{i}") for i in range(3)] + self.assertEqual(positions, sorted(positions)) + self.assertTrue(all(text.count(f"LATE-{i}") == 1 for i in range(3))) + self.assertEqual(self.runtime.generation["id"], generation) + self.assertFalse(any(e["type"] == "generation.finished" for e in self.runtime.events)) + await self.provider.reply("all questions answered") + finished = await self.runtime.wait_for(lambda e: e["type"] == "generation.finished", 3) + self.assertEqual(finished["input_ids"], ["original", "late-0", "late-1", "late-2"]) + self.assertEqual(len([e for e in self.runtime.events if e["type"] == "generation.started"]), 1) + self.assertEqual((await self.runtime.submit_steering(Input("steer", "LATE-1", id="late-1"), generation))["disposition"], "applied") with self.assertRaisesRegex(ValueError, "no longer active"): await self.runtime.submit_steering(Input("steer", "STALE", id="stale"), generation) - self.assertNotIn("stale", self.runtime.accepted) - await self.runtime.submit(Input("user", "EXPLICIT NEXT INPUT", id="next")) + rows = await self.session.coordinator.get("context").get_messages() + self.assertFalse(any(m.get("role") == "user" and m.get("content") == "" for m in rows)) + + async def test_burst_during_model_and_tools_reaches_next_request_once(self): + await self.start() + await self.runtime.submit(Input("user", "ORIGINAL", id="original")) + await self.provider.request() + generation = self.runtime.generation["id"] + await self.runtime.submit_steering(Input("steer", "MODEL-UPDATE", id="model"), generation) + await self.provider.reply(calls=[ToolCall(id="step", name="delegate", arguments={"async": False})]) + await self.runtime.wait_for(lambda e: e["type"] == "tool.pre", 3) + for i in range(3): + await self.runtime.submit_steering(Input("steer", f"TOOL-UPDATE-{i}", id=f"tool-{i}"), generation) + self.tool.release.set() request = await self.provider.request() - self.assertNotIn(command.text, str(request.messages)) - await self.provider.reply("next explicit result") + text = str(request.messages) + updates = ["MODEL-UPDATE", *[f"TOOL-UPDATE-{i}" for i in range(3)]] + positions = [text.index(value) for value in updates] + self.assertEqual(positions, sorted(positions)) + self.assertTrue(all(text.count(value) == 1 for value in updates)) + self.assertEqual(self.tool.calls, 1) + await self.provider.reply("updated answer") + finished = await self.runtime.wait_for(lambda e: e["type"] == "generation.finished", 3) + self.assertEqual(finished["input_ids"], ["original", "model", "tool-0", "tool-1", "tool-2"]) + + async def test_stop_during_final_checkpoint_does_not_resume_pending_steer(self): + await self.start() + await self.runtime.submit(Input("user", "ORIGINAL")) + await self.provider.request() + entered, release = asyncio.Event(), asyncio.Event() + async def checkpoint(): + entered.set() + await release.wait() + self.session.coordinator.register_capability("live.checkpoint", checkpoint) + await self.provider.reply("answer") + await asyncio.wait_for(entered.wait(), 3) + await self.runtime.submit_steering(Input("steer", "NO REPLAY", id="late"), self.runtime.generation["id"]) + await self.runtime.submit(Input("stop", target="cancel")) + release.set() + await asyncio.wait_for(self.task, 3) + self.assertTrue(self.provider.requests.empty()) + self.assertEqual(self.runtime.steering_outcomes["late"]["disposition"], "held") + + async def test_admission_closes_before_terminal_publication_awaits(self): + await self.start() + await self.runtime.submit(Input("user", "ORIGINAL")) + await self.provider.request() + generation = self.runtime.generation["id"] + context = self.session.coordinator.get("context") + original = context.get_messages + entered, release = asyncio.Event(), asyncio.Event() + async def messages(): + if self.runtime.steering_closed: + entered.set() + await release.wait() + return await original() + context.get_messages = messages + await self.provider.reply("done") + await asyncio.wait_for(entered.wait(), 3) + with self.assertRaisesRegex(ValueError, "finishing"): + await self.runtime.submit_steering(Input("steer", "TOO LATE", id="late"), generation) + self.assertNotIn("late", self.runtime.accepted) + release.set() + await self.runtime.wait_for(lambda e: e["type"] == "generation.finished", 3) + self.assertTrue(self.provider.requests.empty()) + + async def test_exhausted_budget_does_not_resume_checkpoint_steer(self): + await self.start() + await self.runtime.submit(Input("user", "ORIGINAL")) + await self.provider.request() + async def checkpoint(): + self.loop._budget_exhausted = True + await self.runtime.submit_steering(Input("steer", "WAIT", id="late"), self.runtime.generation["id"]) + self.session.coordinator.register_capability("live.checkpoint", checkpoint) + await self.provider.reply("budget summary") + await self.runtime.wait_for(lambda e: e["type"] == "generation.finished", 3) + self.assertEqual(self.runtime.steering_outcomes["late"]["disposition"], "held") + self.assertTrue(self.provider.requests.empty()) async def test_cancellation_holds_only_not_yet_injected_anchored_work(self): await self.start() @@ -188,6 +269,24 @@ async def test_explicit_async_opt_in_and_wait_wakes_on_user_input(self): self.assertEqual(self.tool.calls, 1) await self.provider.reply("accepted correction") + async def test_anchored_input_wakes_job_wait_without_stopping_child(self): + await self.start() + await self.runtime.submit(Input("user", "background task")) + await self.provider.request() + generation = self.runtime.generation['id'] + await self.provider.reply(calls=[ToolCall(id="background", name="delegate", arguments={"async": True})]) + await self.provider.request() + job_id = next(iter(self.loop.jobs)) + await self.provider.reply(calls=[ToolCall(id="wait", name="live_job", arguments={"action": "wait", "job_id": job_id, "timeout": 60})]) + await self.runtime.wait_for(lambda e: e['type'] == 'tool.pre' and e.get('tool') == 'live_job', 3) + await self.runtime.submit_steering(Input('steer', 'NEW GUIDANCE', id='steer'), generation) + request = await self.provider.request() + self.assertIn('NEW GUIDANCE', str(request.messages)) + self.assertIn('input_pending', str(request.messages)) + self.assertFalse(self.loop.jobs[job_id]['task'].done()) + self.assertEqual(self.tool.calls, 1) + await self.provider.reply('guidance received while child runs') + async def test_explicit_sync_overrides_legacy_background_default(self): await self.start() await self.runtime.submit(Input("user", "finite delegate")) From eca45ff9021252ea1c5b79e3349814727167c351 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Tue, 6 Oct 2026 17:20:50 -0700 Subject: [PATCH 2/2] Report bounded compaction failure codes without leaking provider errors --- amplifier_module_loop_live/failures.py | 15 +++++++++++++-- tests/test_failures.py | 16 ++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/amplifier_module_loop_live/failures.py b/amplifier_module_loop_live/failures.py index 5a146d5..9510d06 100644 --- a/amplifier_module_loop_live/failures.py +++ b/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} diff --git a/tests/test_failures.py b/tests/test_failures.py index 6e228c1..2286cbd 100644 --- a/tests/test_failures.py +++ b/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