Agents: steering, a message joins the turn being answered at its next step

docs/agents/README.md

@@ -14,9 +14,14 @@ building it, and how to port it.
 
 ## In brief
 
-- **A conversation is a sequence of turns.** A turn is one prompt and what
-  answers it: its steps (the model's replies, tool calls and results) and its
-  answer. Turns never overlap, so at most one turn has no answer: the latest.
+- **A conversation is a sequence of turns.** A turn is one prompt, the
+  messages steered to it, and what answers them: its steps (the model's
+  replies, tool calls and results) and its answer. Turns never overlap, so
+  at most one turn has no answer: the latest.
+- **A message can join the turn being answered.** Sent to `steer` rather than
+  `chat`, the turn takes it at its next step, once the model's tool calls
+  have their results, and the model's next request has it; otherwise it is
+  answered as a turn of its own, in order. Either way, once.
 - **The turn is the unit of storage.** One row per turn (the transcript),
   and every other item tagged with its turn (the trajectories), in one
   sequence of positions that are the message ids clients see.
@@ -36,27 +41,27 @@ building it, and how to port it.
 
 ```mermaid
 flowchart LR
-  C[Client] -- "chat, stop, signal" --> A
+  C[Client] -- "chat, steer, stop, signal" --> A
   C -- "status, messages, sync, watch" --> A
   subgraph A["Agent (session object), key alice"]
     H["chat: a turn"] --> F["Framework run: model, tools"]
     F -- "steps, as they come" --> T
     H -- "prompt, answer" --> T
     T[("Key's database: turns, items, snapshots, call records")]
     V["views: messages, sync"] -- read --> T
-    X["private: settle, compact"] --> T
+    X["private: settle, compact, steered"] --> T
   end
-  H -- "start, close, publish" --> S[("Companion signals, key assistant:alice: turn, waits, revision")]
+  H -- "start, close, publish" --> S[("Companion signals, key assistant:alice: turn, waits, steered messages, revision")]
   F -- "wait, for a tool or an approval" --> S
   C -- "signal resolves a wait" --> S
-  S -- "settle a dead turn" --> X
+  S -- "settle a dead turn, and steered for each steered message" --> X
 ```
 
 ## The pages
 
 | Page | What it holds |
 | --- | --- |
-| [Turn semantics](turns.md) | what a turn is; its life, its end, attempts, stops, dead turns and `settle`; the turn model; what clients see; the guarantees |
+| [Turn semantics](turns.md) | what a turn is; its life, its end, attempts, stops, dead turns and `settle`; steering; the turn model; what clients see; the guarantees |
 | [The conversation](conversation.md) | the data layout (every table), a worked example, which turns a turn reads, what the model gets (the turn projection, summaries, snapshots), what a turn carries, the call records, the schemas |
 | [The frameworks](frameworks.md) | per framework: what it expects, how we map onto it, and what we learned |
 | [What we learned](learnings.md) | the research (coding agents, frameworks), the decisions and why, the lessons, the bugs a review found, the open questions |
@@ -77,6 +82,7 @@ then the same.
 | [`_conversation.py`](../../python/restate/agents/_conversation.py) | the conversation: its tables, `Conversation`, `Turn`, `answer_once`, `compact`, `settle` |
 | [`_messages.py`](../../python/restate/agents/_messages.py) | the conversation as clients read it (`messages`, `sync`) |
 | [`_signals.py`](../../python/restate/agents/_signals.py) | the companion |
+| [`_steering.py`](../../python/restate/agents/_steering.py) | the messages steered to a turn: `take_steers`, at its steps |
 | [`_object.py`](../../python/restate/agents/_object.py) | the agent object: `chat`, the views, the private handlers |
 | [`_calls.py`](../../python/restate/agents/_calls.py) | the call records: a tool's Restate calls, made once |
 | [`_tools.py`](../../python/restate/agents/_tools.py) | the tools built on Restate (`call`, `send`, `workflow`, `subagent`, `wait_for_signal`), and `wait_for` |

docs/agents/access-patterns.md

@@ -290,6 +290,24 @@ In order of value for effort:
   a real server.
 - **The companion and the call records**, not exercised by this harness:
   their writes per turn, and the `restate_agent_calls` read at every turn.
+- **Steering** adds no request to a turn that nobody steers: the companion
+  answers the `publish` a turn makes at each tool call's start and end with
+  the messages waiting for it (one state read more, of its inbox, which is
+  empty then), and a turn takes them at a step with no call of its own;
+  the `taken` that tells the companion is sent from the storing
+  transaction's outbox. A steered message costs the `steer` view, one call
+  to the companion (an inbox entry, a publish, and the placeholder sent to
+  `steered`), the transaction that stores it (an item and a row of
+  `restate_agent_steers`), and the placeholder's short invocation (open the
+  database, one read by id). A later turn reads the steered messages of the
+  transcript it loads as one range of that table per page; the summary
+  view, one range. Measured on a real server with
+  `scripts/steering_traffic.py` (a turn of one tool call): a turn nobody
+  steers is what it is before steering, 47 journal entries and 5 companion
+  calls (`start`, 3 `publish`, `close`); a turn that takes a steered
+  message, 57 entries and 6 calls (the outbox's `taken`); the message
+  itself, the `steer` view (5 entries), the companion's `steer` (13), and
+  the placeholder that finds it taken (3).
 - **Snapshots and `compact`:** a snapshot replaces the previous one (one
   `DELETE` and one `INSERT` of the whole compacted history), and `compact`
   reads the transcript from the summary's start. Both are occasional, but

docs/agents/conversation.md

@@ -16,30 +16,33 @@ records).
   `assistant`) has its own SQL database, and its conversation is the tables
   below in it. Nothing in it is shared with another key.
 - **A conversation is a sequence of turns,** in the order their messages
-  came. A turn is one prompt and what answers it:
+  came. A turn is one prompt, the messages steered to it, and what answers
+  them:
   - its **row**: the prompt, then, once the turn ends, its status, its answer
     and what it carries to later turns. The rows together are the
     *transcript*;
   - its **steps**: everything else it stored, the model's replies, the tool
-    calls and their results, a message that asked again. A turn's steps are
-    its *trajectory*.
+    calls and their results, a message that asked again, the messages
+    steered to it that it took ([Steering a turn](turns.md#steering-a-turn)).
+    A turn's steps are its *trajectory*.
 - **Turns never overlap**, so at most one turn has no answer: the latest
-  ([One message at a time](turns.md#one-message-at-a-time)).
+  ([One turn at a time](turns.md#one-turn-at-a-time)).
 - **Every row belongs to one turn,** and only the latest turn, while it has
   no answer, is written ([The turn model](turns.md#the-turn-model)).
 - **The model's input is composed from the turns in one place,**
   `Conversation.history()`, in one of two ways: the
   [turn projection](#what-the-model-sees), or
   [history from snapshots](#history-from-snapshots).
 - **Clients read the turns as stored**, in two projections of the same rows:
-  the full view (every step) and the summary view (each turn's message and
-  answer), under the same message ids.
+  the full view (every step) and the summary view (each turn's message, the
+  messages steered to it, and its answer), under the same message ids.
 
 ```mermaid
 erDiagram
   restate_agent_turns ||--o{ restate_agent_items : "its steps (items.turn = turns.n)"
   restate_agent_conversation_start ||--o{ restate_agent_turns : "where they start, and their summary"
   restate_agent_snapshots }o--|| restate_agent_turns : "stored in a turn, or between turns"
+  restate_agent_steers |o--|| restate_agent_items : "a steered message (steers.n = items.n)"
   restate_agent_turns {
     int n PK "position: the prompt's message id"
     int seq "the change it was written at"
@@ -123,6 +126,12 @@ CREATE TABLE restate_agent_snapshots (       -- the latest only, for agents with
     turn INTEGER,
     items TEXT NOT NULL
 );
+
+CREATE TABLE restate_agent_steers (          -- the messages steered to a turn that it took (appended)
+    n INTEGER PRIMARY KEY,                   -- the item's position: the message of the user it is
+    turn INTEGER NOT NULL,                   -- the turn's position
+    id TEXT NOT NULL UNIQUE                  -- the message's id: the `steer` request's
+);
 ```
 
 - **Positions.** The two tables share one sequence of positions (`n`), the
@@ -153,6 +162,13 @@ CREATE TABLE restate_agent_snapshots (       -- the latest only, for agents with
 - **A snapshot** takes no position of its own, so message ids and cursors
   don't depend on it, and it is never shown
   ([History from snapshots](#history-from-snapshots)).
+- **A steered message** is an item of its turn, a message of the user made
+  with `Items.steered` (by default `Items.prompt`), and a row of
+  `restate_agent_steers` at the same position, with its id: unique, so a
+  turn stores it once whatever its attempts do, and the agent's `steered`
+  finds that it was taken with one read. Earlier turns' steered messages
+  are one range of that table's primary key, as positions only grow. A drop
+  deletes those after the reply it drops.
 
 ### Around the conversation
 
@@ -301,17 +317,18 @@ the previous turn whole, to give its graph what closed it.
 
 1. The summary of the earliest turns, if they have one: a user's message
    with it, and the assistant's acknowledgement.
-2. The turns before the window, as their prompt and answer: the latest of
-   them that the agent's `budget` holds (32 000 tokens by default, counted
-   as 4 characters each).
+2. The turns before the window, as their prompt, the messages steered to
+   them, and their answer: the latest of them that the agent's `budget`
+   holds (32 000 tokens by default, counted as 4 characters each).
 3. The window's turns, whole.
 
 The model sees the earlier turns' messages, not their tool calls: those of
 the window only. The `messages` view shows every turn whole, as stored.
 
 This is a contract for every integration on the turn projection: **earlier
-turns reach the model only from their rows** (each as its prompt, then
-`Items.reply(answer, agent)`), so a framework must never depend on an
+turns reach the model only from their rows** (each as its prompt, the
+messages steered to it, then `Items.reply(answer, agent)`), so a framework
+must never depend on an
 earlier turn's steps outside the window. What it needs from them, it
 carries ([What a turn carries](#what-a-turn-carries)).
 
@@ -348,7 +365,8 @@ sequenceDiagram
 
 - **What is summarized:** the turns before the next window, but the latest
   that half the budget holds, which stay as they are, with the summary
-  before. Turns stay stored, and `messages` shows them as before.
+  before: each turn's message, the messages steered to it, and its answer.
+  Turns stay stored, and `messages` shows them as before.
 - **Who summarizes:** the built-in agents of OpenAI, ADK and LangGraph, with
   their own model. An agent you configure yourself only trims the earliest
   turns. A LangGraph graph agent never summarizes: its model reads the
@@ -537,7 +555,8 @@ invocation (the next invocation's first call replaces it):
   [What we learned](learnings.md#decisions)): its schemas are written for new
   databases, with no migration from the layout of the 1.4.x releases, whose
   tables no deployment uses. From the next release on, schema changes are
-  appended migrations.
+  appended migrations: the first is `restate_agent_steers`, a new table,
+  which a view that applies it to its own session creates at no cost.
 - **The database** is the session object's SQL engine (sdk-shared-core's,
   the same for the TypeScript SDK), one per key, stored in Restate's state.
   It refuses an expression more than 100 deep: match many pairs with a list