[igalshilman]

Summary

Steering lets a person add to a request while the agent is still working on it ("use Celsius", "skip Friday"). A message sent to the new steer endpoint joins the turn being answered at its next step, so the model reads it in its very next request. If the turn has no step left, the message is answered as the next turn instead. Either way it is answered exactly once, through crashes, retries, stops and kills.

chat is unchanged, and a turn that nobody steers makes no extra calls.

Turn semantics, before and after

	Before	With steering
A turn is	one prompt, and its answer	one prompt, the messages steered to it that it took, and one answer to all of them
A message sent while a turn runs	waits in the key's queue, and is the next turn	sent to chat: the same. Sent to steer: the running turn takes it at its next step, or else it is the next turn
Order	turns answer messages in the order they came	unchanged: a steered message keeps its place in the queue, unless the running turn takes it first
A turn stopped, failed or killed	its prompt is not answered	the same for the messages it took; those it did not take are answered as turns of their own

A step is the point where the model's reply has a result for each of its tool calls, just before the model's next request. It is the only place a turn takes steered messages, so the model never gets one between a tool call and its result.

What happens to a steered message

The message is sent	It is answered
while the turn has a step ahead (a tool call is running, say)	by the running turn, which takes it at that step
while the model writes the turn's final answer	as the next turn, before any message sent after it
when no turn is running	as a turn of its own, which it starts
to an agent whose turns take none, or past a turn's 16	as a turn of its own, after the running turn
to a turn that then stops, fails, is refused or killed	if the turn took it: not at all, as it ended with the turn; otherwise as a turn of its own

The endpoint answers {id, turn}: the message's id, and the turn that holds it, or null when it will be a turn of its own. Clients should send it with an idempotency key, so a retried request is the same message.

How it works

1. The steer view passes the message to the agent's companion (signals).
2. The companion always queues a private steered invocation for it on the key (idempotency key steer:<id>), so the message keeps its place in line. If the running turn takes steered messages, the companion also holds it in an inbox for that turn.
3. The turn learns of it from publish, which it already calls at each tool call's start and end: the response now lists the messages waiting. At its next step, the framework integration calls take_steers. That stores each message once, by its id, as a user message of the turn, made by the framework's own Items (steered, by default a prompt's item), and returns those items, which the integration puts into the request about to go out. The same transaction sends the companion taken, through its outbox.
4. steered runs once the turn has ended. If a turn stored the message, it does nothing; otherwise it answers it as an ordinary turn.

Exactly once comes from two unique keys: one steered invocation per message id (Restate's idempotency), and at most one stored copy per id (a unique column of the new restate_agent_steers table), which steered checks.

The conversation owns which messages were steered: it stores them as the framework marks them, so history(), a retried attempt and the next turns hand them back as they were, and no integration looks them up again.

Taken at the next step

sequenceDiagram
  autonumber
  participant C as Client
  participant V as Agent (steer view)
  participant S as Companion
  participant A as Agent (chat)
  participant P as Agent (steered)
  participant DB as Key's database
  C->>A: chat/send("Plan my week")
  A->>S: start(turn, steers)
  A->>DB: the model's reply, a tool call
  C->>V: steer("Skip Friday")
  V->>S: steer(id, text)
  S-)P: steered(id, text): queued now, behind the turn
  Note over S: inbox: the message, for this turn
  A->>S: publish (the tool call ends)
  S-->>A: revision, steering: [the message]
  A->>DB: the tool's result
  A->>DB: the message, a step of the turn, and taken(id) in the transaction's outbox
  DB--)S: taken(id), once the transaction commits
  A->>A: the model's next request has it
  A->>DB: the answer
  A->>S: close(turn)
  Note over P: runs once the turn ended: the message is stored, so it does nothing

Loading

Too late: the next turn

The model is already writing its final answer, so no step follows. The turn ends without the message, and the message's own turn answers it next, before a chat sent after it.

sequenceDiagram
  autonumber
  participant C as Client
  participant S as Companion
  participant A as Agent (chat)
  participant P as Agent (steered)
  participant DB as Key's database
  C->>A: chat/send("Plan my week")
  Note over A: the tools are done: the model writes the final answer
  C->>S: steer("Skip Friday"), through the steer view
  S-)P: steered(id, text): queued behind the turn
  Note over S: inbox: the message, for this turn
  C->>A: chat/send("And next week?"): queued behind steered
  A->>DB: the answer
  A->>S: close(turn): the message it did not take is forgotten here
  P->>DB: stored by a turn? no
  P->>DB: a turn of its own: "Skip Friday", and its answer
  Note over A: then "And next week?" is answered

Loading

The turn is stopped

A message the turn took is its own, like its prompt, so it is stopped with it. A message the turn had not taken yet is answered as a turn of its own afterwards. A turn that is stopped while it waits to store a message takes none.

sequenceDiagram
  autonumber
  participant C as Client
  participant S as Companion
  participant A as Agent (chat)
  participant P as Agent (steered)
  participant DB as Key's database
  C->>S: steer("Use Celsius"), through the steer view
  S-)P: steered(1): queued behind the turn
  A->>DB: at a step: "Use Celsius", a step of the turn, and taken(1)
  C->>S: steer("And the rain")
  S-)P: steered(2): queued behind steered(1)
  C->>A: stop
  A->>DB: the turn ends cancelled, and "Use Celsius" with it
  A->>S: close(turn): message 2, not taken, is forgotten here
  P->>DB: steered(1): stored by a turn? yes: nothing to do
  P->>DB: steered(2): stored? no: a turn of its own, "And the rain", and its answer

Loading

A crash after the message is stored

The message and taken commit together, so a retried attempt finds the message in the turn, gives it to the model in place, and never takes it again.

sequenceDiagram
  autonumber
  participant A1 as Attempt 1
  participant A2 as Attempt 2
  participant DB as Key's database
  participant S as Companion
  participant P as Agent (steered)
  A1->>DB: at a step: the message and taken(id), in one transaction
  Note over A1: the process dies before the model's request
  DB--)S: taken(id), from the outbox: sent again after a crash, with the same key
  A2->>DB: load: the turn has the message, after the tool's result
  Note over A2: the model's next request has it, and it is never taken again
  A2->>DB: the answer
  A2->>S: close(turn)
  P->>DB: stored by a turn? yes: nothing to do

Loading

What the model and clients see

• The model. In the running turn, a steered message sits where it was taken: after the tool results, before the next request. Later turns see an earlier turn as its prompt, the messages steered to it, then its answer, in history(), the budget and summaries.
• The turn's ask. latest_ask skips steered messages, so the output re-ask, ADK's resumed invocation and a graph's run behave as before.
• Clients. Both messages views show steered messages under their own ids, and status.steering lists those still waiting. AgentClient.steer and /steer in the terminal chat send one.

Frameworks

Each integration takes the messages only as a model request goes out, and puts the items take_steers returns into that request:

• OpenAI: the input filter takes them at each step. The SDK builds each request's input afresh, so the run (_Steers) splices them back into every later request, after the item each came after. A run also takes the messages waiting as it starts. Input guardrails don't run again. An agent used as a tool (Agent.as_tool) runs with the run's run_config, our filter with it: its requests get none of the messages.
• Pydantic AI: an innermost _Steering capability, wrapped by _Snapshots, hands them to Pydantic AI's own RunContext.enqueue(priority='asap'). A snapshot stored by the same request covers only what came before them.
• ADK: _Restate.before_model_callback, for an LlmAgent root, adds them to the request and to the session. Each has an invocation id of its own (_Items.steered), so a retry still resumes the prompt's invocation. The agent's own before_model_callback runs afterwards on the request with them, so a response it gives in place of the model's answers them too.
• LangGraph's built-in agent: _Items.steered gives each the id steer:<id>, which is stored with it, and the model node keeps them in the run's state. A trimmed request never starts at one.

Limits

• 16 per turn. A turn takes at most 16 steered messages in all (STEERS); the companion counts the ones it took until the turn ends. Past that, each is a turn of its own.
• Only at a step. A tool that waits a day for an approval holds the message until it returns; stop the turn to change course sooner.
• Only as a request goes out. A message that no model request follows is a turn of its own. ADK, for one, asks the model nothing more after some tools (skip_summarization, set_model_response).
• Agents that take none, so each steered message is a turn of its own:
  • a LangGraph graph= agent;
  • an ADK agent with a workflow root;
  • a background agent;
  • an OpenAI run after a handoff whose input_filter drops the handoff's own output (see below).

Traffic

Measured on a real server with scripts/steering_traffic.py, for a turn of one tool call:

	Journal entries	Companion calls
A turn nobody steers, on main and on this branch	47	5 (start, 3 × publish, close)
A turn that takes a steered message	57	6 (+ taken, from the outbox)
The message itself	steer view: 5, companion steer: 13, placeholder that finds it taken: 3

When nobody steers, a turn makes no extra calls. Inside the companion, publish does one more state read, of the empty inbox.

Compatibility

• Wire format: additive only (Started.steers, publish's steering, Status.steering). A TypeScript companion has no steer handler, so the steer view fails against it, and turns take nothing from it. The port is step 8 of docs/agents/porting.md.
• Reserved names: steer and steered are now built-in handler names. A user handler with either name is refused.
• Schema: one appended migration, a new table.

Testing

45 new tests, and 13 durability mutants in scripts/resurrect.py (all caught).

• In-process (tests/agents_steering.py, 11 tests): stored once, after a crash too; nothing taken while a call is open, outside a chat turn, or once the turn stopped; earlier turns as prompt, steered messages, answer; the budget counts them at once; the output check still asks again once; a drop frees the message; the summary view and cursors; the migration.
• Companion, real server (tests/agents_signals.py, 3 new): the inbox and the placeholder; publish and taken; a closed turn forgets its messages; clear; the cap, which counts messages already taken.
• Agent with a scripted framework, real server (tests/agents_steering_server.py, 5 tests): taken at the next step; too late, answered next in order; no turn running; an agent that takes none; a killed turn.
• Per framework (26 tests): each checks the message reaches the next request once, and that a crash after it was stored retries with it once. Also:
  • OpenAI (9): a message during the answer; its place in every later request, also after a handoff filter; before the first request; no input guardrail runs again; an agent used as a tool gets none of them, and the turn stores none of its items.
  • Pydantic AI (6): a message during the answer; a snapshot in the same request; feedback and a message through a retry; prompt without snapshots.
  • ADK (6): a message during the answer; a workflow root; a step after which ADK asks nothing; a fresh invocation that a retry resumes past.
  • LangGraph (5): the message stays after the next round, and keeps its id through a retry; a request never starts at it; a graph= agent takes none.
• Full suite, with main merged in: 607 passed. ruff, pyright (strict) and mypy are clean.

Found on the way

• Fixed here:
  • Pydantic AI stored a RetryFeedbackPart request again whenever a run merged or resumed it, on a retry and then on every later turn. _Restate now recognizes feedback by its parts.
  • An OpenAI agent used as a tool (Agent.as_tool) runs with the outer run's run_config, so the turn's input filter stored its tool outputs, without their calls, which broke the next turn's request. A request made while one of the run's tool calls runs is now the nested run's, and gets the user's own filter only.
• Not fixed here, documented in frameworks.md:
  • An OpenAI handoff whose input_filter drops the handoff's own output (remove_all_tools) never stores that output. The next turns' requests then carry a function call with no output, which OpenAI's APIs refuse. Steering takes nothing after such a handoff.
  • What other Pydantic AI capabilities enqueue (the harness's system reminders, loop detection, and so on) reaches the model but is never stored, so a retry and the next turns lack it.

Docs

turns.md is the reference: a turn is now defined as its prompt, the messages steered to it, and their answer, through its life, its end, retries and stops, and Steering a turn has the cases above. Also frameworks.md (a row, Items.steered, and notes per framework), conversation.md, README.md, learnings.md (D6 to D10), porting.md, access-patterns.md and testing.md.

🤖 Generated with Claude Code

[igalshilman]

@pullfrog review

[pullfrog]

ℹ️ No issues in the steering logic. One stale docstring reference, inline.

Reviewed changes

I reviewed the whole steering feature: the companion's inbox and placeholder protocol, how the conversation stores steered messages and reads them back, and each framework integration that takes them at a step.

• Companion protocol (_signals.py): steer always sends the steered placeholder with the idempotency key steer:<id>. It holds the message in a per-turn inbox capped at STEERS, and taken marks it taken. publish returns steering. _closed/reset forget the inbox. I traced the races (steer around a turn's close, taken arriving after the close, a turn retried with the same invocation id). Each message ends up answered exactly once.
• Storage (_conversation.py): Turn.steer mirrors _insert. It stores each message once, using the UNIQUE id of the new restate_agent_steers table, and sends taken from the same transaction's outbox. It is shielded, and checks going() again once it holds the ledger lock. AttemptEnded also sets stop.stopped, so a store waiting on the lock when an attempt ends takes nothing.
• Reading back (_conversation.py, _messages.py): earlier turns carry steers, which feed history(), the budget, summaries and the summary view. The page reads cover the right ranges, and last_of derives seq from the tables, so the cursor moves forward for steered rows.
• latest_ask skips steered messages, so the output re-ask, ADK's resumed invocation and graph runs work as before.
• Framework integrations:
  • OpenAI: the input filter splices the messages back into each later request, and the run also takes them as it starts.
  • Pydantic AI: an innermost _Steering wrapped by _Snapshots calls RunContext.enqueue(priority='asap'). I checked this signature and its semantics against the installed pydantic_ai 2.66.
  • ADK: _Restate is now added per run, and before_model_callback adds the messages to the request and to the session.
  • LangGraph: the model node marks the messages with steer: ids, and _latest never starts a trimmed request at one.
• Other drop callers: the OpenAI drop call sites (_resume, the output-guardrail re-ask, a blocked reply) only drop replies that cannot have steered messages after them. So the deleted restate_agent_steers rows and the companion's taken marks never disagree in practice.

  ^{｜ Fix all ➔ ｜ Fix 👍s ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

[pullfrog]

ℹ️ I found no defects in the core design. Two framework paths were checked less deeply than the rest (see below), and CI is still pending.

Reviewed changes

I reviewed all 36 files at 3104124: the core steering mechanism, the four framework integrations, the summary and transcript changes, the docs, and the tests and mutants.

• steer view and private steered handler (_object.py): the view sends the message to the companion. The companion always queues a steered placeholder for it (idempotency key steer:<id>), so the message keeps its place in the key's queue. The placeholder does nothing if a turn already stored the message.
• Companion inbox (_signals.py): the inbox holds messages only for a turn that started with steers: true, at most STEERS = 16 per turn, counting messages already taken. publish now returns the messages still waiting, taken removes their text, and _closed and reset clear the inbox.
• take_steers and Turn.steer (_steering.py, _conversation.py): these take messages only at a step with no open calls, while the turn is still going. The store is shielded and runs under the ledger lock. The DB uniqueness check and the step and steer rows commit together with the outbox taken send. A new appended table, restate_agent_steers, has a unique id.
• History and views: latest_ask skips steered messages. history(), the budget (_turn_size), summaries and the summary view now include each earlier turn's steered messages. drop deletes steered messages after the dropped reply.
• Frameworks:
  • OpenAI: the input filter takes messages at each step, and _Steers/_spliced put them back into every later request.
  • Pydantic AI: _Steering calls enqueue('asap'). Feedback is now deduplicated per RetryFeedbackPart.
  • ADK: the before_model_callback plugin takes messages, for an LlmAgent root only.
  • LangGraph: the built-in model node takes them, using steer:<id> ids, and a trimmed request never starts at a steered message.

ℹ️ What I checked myself, and what I checked less deeply

I traced the exactly-once path through each of these cases and found no case where a message is answered twice or never:

• the turn crashes after the store (held, plus the unique DB read on retry);
• the turn is stopped while waiting for the ledger (going() is checked again under the lock);
• the turn is killed (the next start closes it and the placeholder answers the message);
• the turn closes before taken arrives (taken is a no-op);
• the view runs again (the companion call rejoins through its idempotency key);
• the cap is reached;
• drop runs after a take (the message is either taken again under the same taken:<id> key or answered by the placeholder).

final_reply() also returns None once a steered message comes last, so a retried turn asks the model again instead of returning a stale answer.

I checked these less deeply: whether _spliced places messages correctly in OpenAI requests after handoffs and user call_model_input_filters, and whether Pydantic AI's enqueue and ADK's plugin ordering behave as the code assumes in the installed versions. I confirmed that RunContext.enqueue exists in the installed Pydantic AI. Beyond that, these rely on the PR's own framework tests.

Local runs:

• tests/agents_steering.py: 11 passed.
• In-process tests matching -k steer across the framework test files: 13 passed.
• agents_conversation, agents_messages and agents_snapshots: 57 passed.
• The framework files, run in-process: no failures. Their Docker-harness tests could not run here.
• ruff and pyright on the core files: clean.

The CI Lint and Test and integration jobs were still pending when I reviewed.

  ^{｜ Fix it ➔ ｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}

[pullfrog]

Follow-up to my review: I've now finished checking how OpenAI places steered messages in later requests (_Steers / _spliced). It holds for parallel tool calls, multi-step runs, handoffs with and without an input_filter, a user's own filter, compaction, _resume and output-guardrail re-asks. One narrow case does not.

A nested Agent.as_tool run gets the outer run's steered messages. When an agent is used as a tool, the SDK passes the outer run_config to the nested Runner.run if none is given (agents/agent.py: resolved_run_config = context.run_config). The nested run's requests then also go through _Hooks.model_input. Once the outer run has taken a message, self.steers.placed(items) (python/restate/agents/openai/_agent.py:329) adds it to every request of the sub-agent, for example [sub prompt, S1, fc z, fo z]. The request is still valid, but the sub-agent sees user messages from the outer conversation, and its item keys are added to self.steers.seen.

Taking messages is not affected: take_steers returns [] while the outer as_tool call is open.

A possible fix: only splice into requests whose input contains this run's own anchor or first item, or whose data.agent belongs to this run. A test with an as_tool sub-agent would pin it.

The other open question from my review, how Pydantic AI, ADK and LangGraph behave in their installed versions, also checked out with no defects.

  ^{｜ View workflow run ｜ via Pullfrog ｜ Using claude-opus-5-5 ｜ 𝕏}

[igalshilman]

Thanks, the as_tool case was real, and it went one step further: the nested run's tool outputs also went through new_results, so the turn stored them without their calls. That breaks the next turn's request.

Fixed in 19a42c5. The run's hooks now count its running tool calls. The SDK makes the run's own next request only once all of its calls have ended, so a request made while one of them runs belongs to a nested run. Such a request gets the user's own filter only: nothing is stored, and no steered message is taken or placed. The fix comes with the test test_an_agent_used_as_a_tool_gets_none_of_the_turns_messages_and_stores_none_of_its_items and the mutant openai-nested-run-filtered, which the test catches.

---

Generated by Claude Code

[pullfrog]

✅ No new issues found. The new commit fixes the Agent.as_tool defect raised in the follow-up comment.

Reviewed changes

I reviewed the one commit since 3104124, which keeps a nested OpenAI run (an agent used as a tool) separate from the turn.

• _Hooks.running gate: on_tool_start and on_tool_end count the run's open tool calls. While any call is open, model_input applies only the user's own call_model_input_filter (_filtered). A nested run therefore gets none of the turn's steered messages, and the turn stores none of the nested run's tool outputs. I checked this against the installed agents SDK. as_tool passes on run_config (and our filter with it), but uses its own hooks argument rather than the outer run's, so the nested run's on_llm_end and on_tool_* never reach _Hooks. A function tool's on_tool_end follows every on_tool_start except in two cases: a nested interruption, which fails the turn here (APPROVALS), and an error the tool raises, which fails the run. So the counter cannot be left non-zero for a later step.
• _filtered extraction: this moved the user-filter call into a helper. The outer path is unchanged.
• Test, mutant and docs: added test_an_agent_used_as_a_tool_gets_none_of_the_turns_messages_and_stores_none_of_its_items, which passes locally, and the openai-nested-run-filtered mutant. The module docstring and frameworks.md now describe the nested run's behaviour.

One behaviour change, which I consider acceptable: a nested run's request no longer sets requesting. If a nested model error propagates out of the tool, _go now classifies it as a tool's error (from_tool) rather than a provider refusal. That matches where the error happened from the turn's point of view.

  ^{｜ View workflow run ｜ Using claude-opus-5-5 ｜ 𝕏}