From 70e4b0b8fbdfe6d1b2d45af382316a8c7b005062 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Fri, 2 Oct 2026 15:48:33 -0700 Subject: [PATCH] Add opt-in generation-anchored steering with explicit disposition --- README.md | 2 +- amplifier_module_loop_live/orchestrator.py | 56 +++++++++++-- amplifier_module_loop_live/provenance.py | 3 +- amplifier_module_loop_live/runtime.py | 39 ++++++++- docs/anchored-steering.md | 47 +++++++++++ tests/test_loop.py | 93 ++++++++++++++++++++++ tests/test_runtime.py | 16 ++++ 7 files changed, 247 insertions(+), 9 deletions(-) create mode 100644 docs/anchored-steering.md diff --git a/README.md b/README.md index bf097cd..045d354 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ requires reconciliation, even if the same result was announced earlier. Live responses carry generation and delivered-input identities, separate from provisional assistant content and native steering acceptance. See the -[generation event contract](docs/generation-events.md) before using completion +[generation event contract](docs/generation-events.md) and [anchored steering](docs/anchored-steering.md) before using completion events to resolve a host request or send a notification. Hosts sharing session storage may supply optional `live.activation` and diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py index 4460374..c14aae5 100644 --- a/amplifier_module_loop_live/orchestrator.py +++ b/amplifier_module_loop_live/orchestrator.py @@ -65,6 +65,16 @@ async def _execute_one_turn(self, *args, **kwargs): def _select_provider(self, providers): return self.root_provider or super()._select_provider(providers) + async def _hold_anchored(self, reason): + retained = deque() + for command in self.pending: + if command.target_generation_id is None: + retained.append(command) + else: + await self.runtime.emit("steering.held", input_id=command.id, + target_generation_id=command.target_generation_id, reason=reason) + self.pending = retained + async def _drain_steering(self, context, hooks, iteration): if self.runtime is None: return await super()._drain_steering(context, hooks, iteration) @@ -74,18 +84,28 @@ async def _drain_steering(self, context, hooks, iteration): legacy = [text for text in self._steering_queue.drain() if text != _WAKE] commands = list(self.pending) self.pending.clear() + applied = [] for command in commands: + if command.target_generation_id is not None and (not self.runtime.generation or self.runtime.generation["id"] != command.target_generation_id): + await self.runtime.emit("steering.held", input_id=command.id, + target_generation_id=command.target_generation_id, reason="generation-ended") + continue content = self.host.content(command, self.coordinator) if command.attachments else self._text(command) await context.add_message({"role": "user", "content": content, "metadata": input_metadata(command)}) await self.runtime.emit("input.delivered", input_id=command.id, - delivery="request_boundary", source=command.source) + delivery="request_boundary", source=command.source, + **({"target_generation_id": command.target_generation_id} if command.target_generation_id else {})) + if command.target_generation_id is not None: + await self.runtime.emit("steering.applied", input_id=command.id, + target_generation_id=command.target_generation_id, delivery="request_boundary") + applied.append(command) for text in legacy: await context.add_message({"role": "user", "content": text}) - if commands or legacy: + if applied or legacy: await hooks.emit("orchestrator:steering_injected", { "orchestrator": "loop-live", "iteration": iteration, - "input_ids": [c.id for c in commands], "queued_remaining": 0}) - return len(commands) + len(legacy) + "input_ids": [c.id for c in applied], "queued_remaining": 0}) + return len(applied) + len(legacy) def native_job(self, call_id): return next((job for job in self.jobs.values() if job["call_id"] == call_id), None) @@ -159,6 +179,10 @@ 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.anchored_steering_mode = "unavailable" if native else "request_boundary" + coordinator.register_capability("live.steering", {"version": 1, + "mode": runtime.anchored_steering_mode, "cancellable": False, + "submit": runtime.submit_steering}) active = None last = "" last_activation = None @@ -229,6 +253,8 @@ async def turn(command): if prompt: self.pending.append(Input("user", prompt)) while True: + if active is None and self.pending: + await self._hold_anchored("generation-ended") if active is None and self.pending and runtime.inbox.empty(): command = self.pending.popleft() # Publish busy state before scheduling: a host control can @@ -277,6 +303,13 @@ async def turn(command): if value.kind == "service": await runtime.emit("observation.received", input_id=value.id, source=value.source, text=value.text) + if value.target_generation_id is not None: + if active is None or not runtime.generation or runtime.generation["id"] != value.target_generation_id: + await runtime.emit("steering.held", input_id=value.id, + target_generation_id=value.target_generation_id, reason="generation-ended") + continue + await runtime.emit("steering.accepted", input_id=value.id, + target_generation_id=value.target_generation_id, delivery="request_boundary") # Multimodal updates enter through the normal Amplifier # message serializer at the next request boundary. if native and not value.attachments and await native.steer_live(value): @@ -288,6 +321,7 @@ async def turn(command): await active active = None result, error = value + await self._hold_anchored("generation-failed" if error else "generation-completed") if error: cause, failure = error await runtime.emit("generation.failed", **failure) @@ -331,6 +365,8 @@ async def turn(command): if active: active.cancel() await asyncio.gather(active, return_exceptions=True) + await self._hold_anchored("session-closing") + if active: await runtime.emit("generation.detached", cancellation="unconfirmed") for job in self.jobs.values(): if not job["task"].done(): @@ -341,12 +377,21 @@ async def turn(command): await provider.close_live() while not runtime.inbox.empty(): kind, value = runtime.inbox.get_nowait() - if kind == "bundle_job": + if kind == "input": + runtime.queued_inputs -= 1 + if value.target_generation_id is not None: + await runtime.emit("steering.held", input_id=value.id, + target_generation_id=value.target_generation_id, reason="session-closing") + elif kind == "bundle_job": await self._job_returned(*value, deliver=False) elif kind == "child_event": await runtime.emit("child.updated", **value) elif kind == "bundle_persistence_error": await runtime.emit("persistence.failed", error_type=value) + for identity, outcome in list(runtime.steering_outcomes.items()): + if outcome["disposition"] == "queued": + await runtime.emit("steering.unknown", input_id=identity, + target_generation_id=outcome["generationId"], reason="injection-disposition-unconfirmed") for remove in unregister: if callable(remove): remove() @@ -359,6 +404,7 @@ async def turn(command): except Exception as exc: await runtime.emit("persistence.failed", error_type=type(exc).__name__) await runtime.emit("session.closed", status=status) + runtime.anchored_steering_mode = None self.running, self.runtime = False, None return json.dumps({"worker_status":status,"last_report":last,"effects":"not_rolled_back"}) if coordinator.get_capability("live.child") and status!="completed" else last diff --git a/amplifier_module_loop_live/provenance.py b/amplifier_module_loop_live/provenance.py index c9f35c2..5301451 100644 --- a/amplifier_module_loop_live/provenance.py +++ b/amplifier_module_loop_live/provenance.py @@ -3,7 +3,8 @@ def input_metadata(command): return {'amplifier_input': {'version': 1, 'kind': command.kind, 'id': command.id, 'source': command.source, - **({'call_id': command.call_id} if command.call_id else {})}} + **({'call_id': command.call_id} if command.call_id else {}), + **({'target_generation_id': command.target_generation_id} if command.target_generation_id else {})}} class InputContext: diff --git a/amplifier_module_loop_live/runtime.py b/amplifier_module_loop_live/runtime.py index d307675..40ab478 100644 --- a/amplifier_module_loop_live/runtime.py +++ b/amplifier_module_loop_live/runtime.py @@ -5,7 +5,7 @@ import json import time import uuid -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace @dataclass(frozen=True) @@ -20,6 +20,8 @@ class Input: # Host-private capability. It is never serialized into context or exposed # to providers; the loop binds it only while executing this input. activation: object | None = field(default=None, compare=False, repr=False) + # Opt-in: this steering input may affect only the named generation. + target_generation_id: str | None = None class _Event(tuple): @@ -60,6 +62,8 @@ def __init__(self, session_id=None, observer=None, max_input_chars=16000): self.closed = False self.max_input_chars = max_input_chars self.generation = None + self.anchored_steering_mode = None + self.steering_outcomes = {} async def submit(self, command: Input): if self.closed: @@ -83,6 +87,13 @@ async def submit(self, command: Input): if previous != command: raise ValueError("Command identity reused with different content") return command.id + 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") + if self.anchored_steering_mode != "request_boundary": + 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") # 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") @@ -91,15 +102,30 @@ async def submit(self, command: Input): self.inbox.put_nowait(("input", command)) self.queued_inputs += 1 self.accepted[command.id] = command + 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) return command.id + async def submit_steering(self, command: Input, generation_id: str): + """Opt-in current-generation delivery; never starts a later turn. + + Acceptance only proves queue admission. Applied/held disposition is + observed separately; the application owns durable command receipts. + """ + if command.kind != "steer" or not isinstance(generation_id, str) or not 1 <= len(generation_id) <= 128: + raise ValueError("submit_steering requires a steering input and bounded generation identity") + if command.target_generation_id is not None and command.target_generation_id != generation_id: + raise ValueError("Conflicting generation anchors") + await self.submit(replace(command, target_generation_id=generation_id)) + return copy.deepcopy(self.steering_outcomes[command.id]) + 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.generation = {"id": data["generation_id"], "input_ids": [], "accepted_input_ids": []} + 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: if event_type in {"input.delivered", "steering.applied"}: @@ -114,6 +140,15 @@ async def emit(self, event_type, **data): data["generation_id"] = generation["id"] data["input_ids"] = list(generation["input_ids"]) data["accepted_input_ids"] = [identity for identity in generation["accepted_input_ids"] if identity not in generation["input_ids"]] + identity = data.get("input_id") + outcome = self.steering_outcomes.get(identity) + if outcome is not None and data.get("target_generation_id") == outcome["generationId"]: + if event_type in {"input.delivered", "steering.applied"}: + self.steering_outcomes[identity] = {**outcome, "disposition": "applied"} + elif event_type == "steering.held" and outcome["disposition"] != "applied": + self.steering_outcomes[identity] = {**outcome, "disposition": "held", "reason": data.get("reason", "generation-ended")} + elif event_type == "steering.unknown" and outcome["disposition"] == "queued": + self.steering_outcomes[identity] = {**outcome, "disposition": "unknown", "reason": data.get("reason", "injection-unconfirmed")} self.sequence += 1 event = {"version": 1, "sequence": self.sequence, "session_id": self.session_id, "time": time.time(), diff --git a/docs/anchored-steering.md b/docs/anchored-steering.md new file mode 100644 index 0000000..7a3d910 --- /dev/null +++ b/docs/anchored-steering.md @@ -0,0 +1,47 @@ +# Steering one active generation + +Hosts that must never turn late steering into a new request can opt into the +`live.steering` coordinator capability. Ordinary `Input("steer", ...)` keeps its +existing behavior; this contract is additive and requires a live runtime. + +The capability is `{version: 1, mode: "request_boundary" | "unavailable", +cancellable: false, submit: runtime.submit_steering}`. A host captures the active +`runtime.generation["id"]`, then calls: + +```python +receipt = await capability["submit"]( + Input("steer", "Use the corrected path", id=stable_input_id), generation_id +) +``` + +The receipt contains `accepted`, `inputId`, `generationId`, and `disposition`. +`queued` proves admission only. `applied` proves successful insertion into the +active generation's context; it does not prove a provider read or complied with +the correction. `held` proves the input was not injected and will not start a +later turn. `unknown` means context mutation was interrupted and its effect +cannot be confirmed. Held and unknown input must never be replayed implicitly. +The host owns durable receipts across process loss. The runtime retains at most +2,000 accepted identities and refuses new work when full rather than evicting +identity evidence. Repeating an identical ID in the same live runtime returns +its existing disposition; changing its content or anchor is refused. + +Successful context insertion emits `input.delivered` and `steering.applied`, +with `input_id`, `target_generation_id`, and `delivery: "request_boundary"`. +`steering.held` and `steering.unknown` include the same input/target identities and +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. + +Provider-native `steer_live` is deliberately unavailable through this contract +until it can supply a matching attribution and withdrawal/effect protocol. +There is no replace/retract API: cancellation must not claim to undo context +already written. A cancelled request-boundary context write reports unknown if +its completion cannot be established. + +Tests use actual Amplifier Core/context/loop modules with deterministic provider +and tool fixtures. They cover normal application, late arrival during checkpoint, +cancellation before injection, interruption after a partial context effect, +identity reuse, and unsupported/stale generation rejection. They make no live +provider or account acceptance claim. diff --git a/tests/test_loop.py b/tests/test_loop.py index f32e0ef..4672a18 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -48,6 +48,99 @@ async def execute(self, input): return ToolResult(success=True, output={"report": "CHILD-RESULT"}) class BundleTests(unittest.IsolatedAsyncioTestCase): + async def test_anchored_steering_applies_only_to_its_current_generation(self): + await self.start() + await self.runtime.submit(Input("user", "ORIGINAL", id="original")) + await self.provider.request() + generation = self.runtime.generation["id"] + capability = self.session.coordinator.get_capability("live.steering") + self.assertEqual((capability["version"], capability["mode"], capability["cancellable"]), (1, "request_boundary", False)) + result = await capability["submit"](Input("steer", "ANCHORED", id="anchored"), generation) + self.assertEqual(result["disposition"], "queued") + await self.runtime.wait_for(lambda e: e["type"] == "input.queued" and e["input_id"] == "anchored", 3) + self.tool.release.set() + await self.provider.reply(calls=[ToolCall(id="boundary", name="delegate", arguments={"async": False})]) + request = await self.provider.request() + self.assertIn("ANCHORED", str(request.messages)) + applied = await self.runtime.wait_for(lambda e: e["type"] == "steering.applied" and e["input_id"] == "anchored", 3) + self.assertEqual(applied["target_generation_id"], generation) + duplicate = await capability["submit"](Input("steer", "ANCHORED", id="anchored"), generation) + self.assertEqual(duplicate["disposition"], "applied") + rows = await self.session.coordinator.get("context").get_messages() + matching = [row for row in rows if row.get("content") == "ANCHORED"] + self.assertEqual(len(matching), 1) + self.assertEqual(matching[0]["metadata"]["amplifier_input"]["target_generation_id"], generation) + await self.provider.reply("done") + 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): + 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 asyncio.wait_for(checkpoint_entered.wait(), 3) + await self.runtime.submit_steering(command, generation) + 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") + 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")) + request = await self.provider.request() + self.assertNotIn(command.text, str(request.messages)) + await self.provider.reply("next explicit result") + + async def test_cancellation_holds_only_not_yet_injected_anchored_work(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", "HELD ON CANCEL", id="held"), generation) + await self.runtime.wait_for(lambda e: e["type"] == "input.queued" and e["input_id"] == "held", 3) + await self.runtime.submit(Input("stop", target="cancel")) + await asyncio.wait_for(self.task, 3) + self.assertEqual(self.runtime.steering_outcomes["held"]["disposition"], "held") + self.assertNotIn("HELD ON CANCEL", str(await self.session.coordinator.get("context").get_messages())) + self.assertTrue(self.provider.requests.empty()) + + async def test_interrupted_context_injection_is_unknown_not_false_held_evidence(self): + await self.start() + context = self.session.coordinator.get("context") + original = context.add_message + injected, release = asyncio.Event(), asyncio.Event() + async def add(message): + await original(message) + if message.get("metadata", {}).get("amplifier_input", {}).get("id") == "uncertain": + injected.set() + await release.wait() + context.add_message = add + await self.runtime.submit(Input("user", "ORIGINAL", id="original")) + await self.provider.request() + await self.runtime.submit_steering(Input("steer", "PARTIAL EFFECT", id="uncertain"), self.runtime.generation["id"]) + await self.runtime.wait_for(lambda e: e["type"] == "input.queued" and e["input_id"] == "uncertain", 3) + self.tool.release.set() + await self.provider.reply(calls=[ToolCall(id="boundary", name="delegate", arguments={"async": False})]) + await asyncio.wait_for(injected.wait(), 3) + await self.runtime.submit(Input("stop", target="cancel")) + await asyncio.wait_for(self.task, 3) + context.add_message = original + self.assertEqual(self.runtime.steering_outcomes["uncertain"]["disposition"], "unknown") + self.assertIn("PARTIAL EFFECT", str(await context.get_messages())) + self.assertFalse(any(e["type"] == "steering.held" and e.get("input_id") == "uncertain" for e in self.runtime.events)) + async def test_compaction_accepts_input_and_continues_same_task(self): try: from amplifier_module_context_managed.boundary import mount_boundary diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 0356e40..1210132 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -37,6 +37,22 @@ async def test_closed_runtime_rejects_input(self): with self.assertRaises(RuntimeError): await runtime.submit(Input("user", "late")) + async def test_anchored_steering_refuses_unsupported_and_stale_modes_before_acceptance(self): + runtime = Runtime() + await runtime.emit("generation.started", generation_id="generation", initial_input_id="root") + for mode in (None, "native", "unavailable"): + runtime.anchored_steering_mode = mode + with self.assertRaisesRegex(ValueError, "unavailable"): + await runtime.submit_steering(Input("steer", "do not queue", id="steer"), "generation") + self.assertEqual(runtime.queued_inputs, 0) + self.assertEqual(runtime.accepted, {}) + runtime.anchored_steering_mode = "request_boundary" + with self.assertRaisesRegex(ValueError, "no longer active"): + await runtime.submit_steering(Input("steer", "stale", id="steer"), "old-generation") + with self.assertRaises(ValueError): + await runtime.submit_steering(Input("steer", "missing generation"), None) + self.assertEqual(runtime.accepted, {}) + def test_core_has_no_host_or_provider_imports(self): import ast import amplifier_module_loop_live