Anchor opt-in steering to one generation and preserve uncertain outcomes

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

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
 

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:

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(),

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.