Agents: fewer database commits and companion calls per turn

docs/agents/access-patterns.md

@@ -291,9 +291,10 @@ In order of value for effort:
 - **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;
+  answers the publishes a turn makes anyway (its `start`, and one after each
+  write of steps that commits: a reply with tool calls, the tools' results)
+  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
@@ -303,11 +304,12 @@ In order of value for effort:
   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).
+  steers, 40 journal entries and 4 companion calls (`start`, which
+  publishes; a `publish` after the reply with the call, and one after its
+  result; `close`); a turn that takes a steered message, 50 entries and 5
+  calls (the outbox's `taken`); the message itself, the `steer` view (5
+  entries), the companion's `steer` (12), 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

@@ -524,9 +524,10 @@ invocation (the next invocation's first call replaces it):
   again under the same id; a new call with the same input (the model asking
   again after the first's answer) waits anew, and never gets an earlier
   call's approval or denial.
-- **Forgetting.** At each publish and before each tool call, a record whose
-  Restate call returned is forgotten once the transcript has its result
-  under the latest id it was made with: a success, or the very error the call
+- **Forgetting.** At each publish (after each write of steps that commits)
+  and before each tool call, a record whose Restate call returned is
+  forgotten once the transcript has its result under the latest id it was
+  made with: a success, or the very error the call
   returned. Another error in the transcript (one the framework made up for a
   cut-off call) doesn't count: the call may be made again under a new id, and
   must find its record.
@@ -550,6 +551,9 @@ invocation (the next invocation's first call replaces it):
   discards, so a key never written reads as empty tables. The names are
   `restate_agents.turns`, `restate-agents.calls`, `restate_agents.langgraph`,
   `restate_agents.pydantic_ai` and `restate_agents.pydantic_ai.memory`.
+  A turn (`answer_once`) applies the conversation's and the tool calls'
+  together (`ensure_schemas`), as it reads the call records as it ends,
+  whether or not a tool ran: a key's first turn pays one schema commit.
 - **Migrations.** `restate.sessions`' rule is: never change a migration that
   has run; append one. This release is the exception (D1 in
   [What we learned](learnings.md#decisions)): its schemas are written for new

docs/agents/frameworks.md

@@ -176,33 +176,44 @@ and it leans on the harness wherever a capability does what our code would
 ### The design
 
 **History from snapshots.** The harness's compaction replaces our window,
-budget and summaries. When compaction rewrote the history, `_Restate`, which
-runs innermost, stores the rewritten list as a snapshot of the turn before the
-request goes out; the next turn starts from it, and its input only grows until
-the next compaction. An `Agent` of your own with no compaction capability gets
-the same default (Q3).
+budget and summaries. `_Restate` comes first of the run's capabilities (it
+has no position of its own): before each request it stores the run's new tool
+results, and after it the response. When compaction rewrote the
+history, `_Snapshots`, which runs after every other capability's
+`before_model_request` but `_Steering`'s, stores the rewritten list as a
+snapshot of the turn before the request goes out; the next turn starts from
+it, and its input only grows until the next compaction. `_Steering`, innermost
+inside it, takes the messages steered to the turn last, so that a snapshot
+this request stores comes before them. An `Agent` of your own with no
+compaction capability gets the same default (Q3).
 
 ```mermaid
 sequenceDiagram
   autonumber
   participant R as Run (turn N)
+  participant G as _Restate (first)
   participant T as TieredCompaction
-  participant G as _Restate (innermost)
+  participant S as _Snapshots
+  participant St as _Steering (innermost)
   participant DB as Key's database
   participant M as Model
   R->>DB: load: the latest snapshot, then every row after it, whole
   loop each model request
-    R->>T: before_model_request
+    R->>G: before_model_request
+    G->>DB: store new tool results, by call id
+    G->>T: before_model_request
     opt over the target
       T->>T: clamp, clear old tool results, then summarize if needed
     end
-    T->>G: before_model_request (the messages as they will be sent)
+    T->>S: before_model_request (the messages as they will be sent)
     opt the history was rewritten, not extended
-      G->>DB: store a snapshot of the turn: the compacted history, minus system prompt parts
+      S->>DB: store a snapshot of the turn: the compacted history, minus system prompt parts
     end
-    G->>DB: store new tool results, by call id
-    G->>M: the request
-    M-->>G: the response
+    S->>St: before_model_request
+    St->>DB: store the messages steered to the turn (take_steers), then enqueue them
+    R->>M: the request, with the enqueued messages
+    M-->>R: the response
+    R->>G: after_model_request
     G->>DB: store the response
   end
   Note over DB: turn N+1 loads from this snapshot: its input only grows until the next compaction
@@ -265,7 +276,9 @@ handlers have no `ctx.run`; it disables harness features), `StepPersistence`
   order in the call. Replaying only the writes isn't enough: the call's
   earlier writes are committed, and reading them changes what it writes next
   (which steps a status change unblocks), so its writes no longer line up.
-  This is what the harness's own durable stores assume ("the store's
+  A read's journal row is kept in memory and written with the call's next
+  write: only a write makes it matter, so a call that only reads commits
+  nothing. This is what the harness's own durable stores assume ("the store's
   recorded result keeps the original ids").
 - **A snapshot is the history the model will get,** so: strip system prompt
   parts (or `ReinjectSystemPrompt` stops reinjecting it); drop the input

docs/agents/learnings.md

@@ -79,7 +79,7 @@ Each framework's details are in [The frameworks](frameworks.md).
 | — | **The summary's form stays per framework.** OpenAI's is one user message; ADK and LangGraph keep a user message then "Understood.". | OpenAI's was checked against both its APIs; the others run on scripted models in the tests, which can't tell what Gemini or other chat models accept. |
 | D6 | **Steering is a view of its own, `steer`; `chat` keeps one message per turn.** | A client that sends to `chat` gets the answer to its message as `chat`'s response, which joining another turn would take from it. A UI that wants one way in sends everything to `steer`, which starts a turn when none is answered. |
 | D7 | **A steered message is queued as a turn of its own the moment it is sent** (the agent's `steered`), and the turn that takes it makes that turn do nothing. | Its place in the key's queue is kept with the messages sent after it, a turn that dies leaves no message behind, and the companion forwards nothing at a turn's close. The cost is one short invocation per steered message. |
-| D8 | **The turn learns of steered messages from `publish`**, which it makes at each tool call's start and end, and takes them with an outbox send. | A turn that nobody steers makes no call more: the check is free. A call of its own at each step would cost one round trip per tool round of every turn. |
+| D8 | **The turn learns of steered messages from `publish`**, which it makes anyway (its `start`, and after each write of steps that commits), and takes them with an outbox send. | A turn that nobody steers makes no call more: the check is free. A call of its own at each step would cost one round trip per tool round of every turn. |
 | D9 | **Each framework takes a steered message in place**, in the request about to go out (OpenAI's input filter, Pydantic AI's `enqueue`, ADK's session events, LangGraph's model node); none restarts its run. | Restarting a run from the stored steps (the crash path) reran input guardrails, reset `max_turns`, usage limits and recursion limits, and left the old run able to write. |
 | D10 | **A message taken late is the next turn**: one that comes after the turn's last step is answered next, not folded into the turn. | The turn's answer stays the answer to what it was asked; folding it in would hide that answer from the summary view. |
 | — | **Steps retention is future work** ([its design](steps-retention.md)). | Nothing reads a turn's steps once it leaves the window, but deleting them changes what the full view and conversation search can show; it can come later, as an option. |

docs/agents/porting.md

@@ -57,6 +57,15 @@ Each step stands on its own, with its tests, and the earlier ones are small.
      terms (`checkpointer.ts` builds such `OR`s); and, if the TypeScript
      LangGraph has `DeltaChannel`, keep what it is rebuilt from when pruning
      (`_kept` in `_checkpointer.py`).
+   - **The companion's `start` publishes.** A turn that says so
+     (`Started.publishes`, a followed turn) is published as `start` starts
+     it, and `start` returns what `publish` returns (`{revision,
+     steering?}`): the turn makes no publish of its own to start. A turn that
+     doesn't say so gets nothing back, and is not published: so a TypeScript
+     turn that this SDK's companion holds publishes its start as before, and
+     this SDK's turn publishes its own start when the TypeScript companion
+     returns nothing. `publish` (and `start`) clear `watching` only when a
+     watch waits.
 2. **The companion's `settle`.** A turn's start says whether its agent
    settles (`Started.settles`); when `awaitTurn` finds the invocation gone,
    the companion closes the turn and sends such an agent `settle(turn)`, a

docs/agents/turns.md

@@ -112,16 +112,16 @@ sequenceDiagram
   participant DB as Key's database
   participant S as Companion (signals)
   C->>A: chat("Plan my week")
-  A->>S: start(turn, settles)
+  A->>S: start(turn, settles, publishes)
   S-)S: awaitTurn(turn)
-  Note over S: status.answering = turn
-  A->>S: publish
+  Note over S: status.answering = turn, publish
+  S-->>A: the revision, and the messages steered to the turn, if any
   A->>DB: the prompt's answer? (none yet)
   A->>DB: load the transcript and the window
   A->>DB: ask: the turn's row
   loop each step of the framework
-    A->>DB: append the model's reply, the tool calls' results
-    A->>S: publish (a tool call starts, ends)
+    A->>DB: append the model's reply with tool calls, or the tools' results
+    A->>S: publish (after the commit)
     S-->>A: the revision, and the messages steered to the turn, if any
     opt messages steered to the turn wait, at the step
       A->>DB: take them: steps of the turn, and taken in the outbox
@@ -143,7 +143,11 @@ sequenceDiagram
    It also says the agent has `settle` (see [Dead turns](#dead-turns)). A
    retried attempt calls `start` again with the same invocation, which
    changes nothing (it keeps when the turn started). `start` sends
-   `awaitTurn(turn)`, the turn's watch.
+   `awaitTurn(turn)`, the turn's watch. It also publishes, as the turn's
+   first revision, and returns what `publish` returns: the turn makes no
+   publish of its own to start. A companion that returns nothing from
+   `start` (an older deployment, or the TypeScript SDK's) gets a `publish`
+   from the turn instead.
 2. **The prompt.** `answer_once` reads the prompt's answer by its id, one
    indexed read: a retried attempt of an answered turn returns it, and reads
    nothing more; one of a turn that was closed (cancelled, failed, refused, or
@@ -152,12 +156,15 @@ sequenceDiagram
    ([Which turns a turn loads](conversation.md#which-turns-a-turn-loads)), closes a turn
    left without an answer, and stores the prompt as the turn's row.
 3. **Steps.** The framework answers: model calls and tool calls. Each step
-   is stored as it comes (a reply in one transaction, each result as it
-   comes back), and each tool call publishes as it starts and as it ends,
-   so a follower sees the call start and its result. The companion answers
-   each `publish` with the messages steered to the turn that wait: at its
-   next step, the turn takes them, and the model's next request has them
-   ([Steering a turn](#steering-a-turn)).
+   is stored as it comes: a reply with tool calls in one transaction, before
+   its tools run; the tools' results before the model is asked again. A
+   final reply, which no tool waits for, is held and commits with the
+   answer. After each write that commits, the turn publishes once, and
+   waits for it: so a follower sees the calls as their tools start, and
+   their results before the model's next request returns. The companion
+   answers each `publish` with the messages steered to the turn that wait:
+   at its next step, the turn takes them, and the model's next request has
+   them ([Steering a turn](#steering-a-turn)).
 4. **Answer.** The answer is stored on the turn's row, with what the turn
    carries. If the turns before the next turn's window outgrow the agent's
    budget, the agent sends itself `compact`
@@ -168,8 +175,8 @@ sequenceDiagram
    turn's error.
 
 A **background** agent (`background=True`, a subagent say) publishes its
-turn's end only: no step, nor its start, wakes a follower. Its turns take no
-steered messages.
+turn's end only: no step, nor its start, wakes a follower (its `start` says
+it does not publish). Its turns take no steered messages.
 
 ## How a turn ends
 
@@ -450,9 +457,9 @@ sequenceDiagram
   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->>S: publish (after the result's commit)
+  S-->>A: revision, steering: [the message]
   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
@@ -520,10 +527,12 @@ sequenceDiagram
   queue as it is sent; and, if the turn being answered takes steered
   messages (its `start` says so, `Started.steers`), it holds the message for
   that turn (`inbox`), 16 at most (`STEERS`).
-- **How the turn learns of it.** From `publish`, which the turn makes at
-  each tool call's start and end anyway: the companion answers it with the
-  messages waiting for the turn. A turn that nobody steers makes no call
-  more.
+- **How the turn learns of it.** From `publish`, which the turn makes
+  anyway as it starts (its `start`) and after each write of steps that
+  commits (a reply with tool calls, the tools' results): the companion
+  answers it with the messages waiting for the turn. The turn waits for the
+  publish after the results, and its next step then takes what it returned.
+  A turn that nobody steers makes no call more.
 - **Taken once.** At a step, the framework's integration calls
   `take_steers`: the turn stores each waiting message by its id, once, as a
   message of the user that is a step of the turn, and the integration puts it
@@ -657,8 +666,8 @@ own answer, and the companion's turn is the invocation, all of them.
 
 The revision increases at:
 
-- a turn's start, and each of its tool calls' start and end (not in the
-  background);
+- a turn's start (its `start`), and each write of its steps that commits:
+  a reply with tool calls, the tools' results (not in the background);
 - a turn's close: its own, its watch's, or the next turn's `start` closing
   it;
 - `settle`, once it closed a dead turn in the conversation;
@@ -668,7 +677,9 @@ The revision increases at:
 - `clear`.
 
 So a follower that reads each new revision sees every turn begin, its tool
-calls, its end, and a killed turn end.
+calls as their tools start, their results before the model's next request
+returns, its end, and a killed turn end. A final reply, held until the
+answer, shows with the turn's end.
 
 ## Agents of the TypeScript SDK
 
@@ -684,7 +695,11 @@ does not show), and only such a turn is sent it:
 
 A turn says it takes steered messages the same way (`Started.steers`): the
 companion holds messages only for such a turn, and every other gets them as
-turns of their own. A TypeScript companion, which has no `steer`, answers
+turns of their own. And it says whether `start` publishes
+(`Started.publishes`): this SDK's companion then publishes and returns what
+`publish` returns; a TypeScript companion returns nothing from `start`, and
+this SDK's turn then publishes its start itself. A TypeScript agent's turn
+does not say so, and this SDK's companion publishes nothing at its `start`. A TypeScript companion, which has no `steer`, answers
 `publish` with no `steering`, so this SDK's turns take none from it; and the
 `steer` view fails there, as the companion has no such handler.
 

python/restate/agents/_calls.py

@@ -46,6 +46,7 @@
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass, field
 
+from restate.agents._common import flush_held
 from restate.sessions import SessionContext
 from restate.sessions.internal import Schema, ScopeKey
 
@@ -91,7 +92,12 @@ class _Calls:
 
 
 async def _store(ctx: SessionContext, state: _Calls) -> None:
-    """Store ``state``; it is committed before the call it records is made."""
+    """Store ``state``; it is committed before the call it records is made.
+
+    The rows the conversation holds back are written first: the call outlives the attempt, and what led to it must
+    too.
+    """
+    await flush_held(ctx)
     await ctx.db.execute(
         "INSERT OR REPLACE INTO restate_agent_calls (row, invocation, calls) VALUES (0, ?, ?)",
         [state.invocation, json.dumps(state.calls)],

python/restate/agents/_common.py

@@ -65,6 +65,27 @@ def next_prompt_id(ctx: SessionContext) -> str:
     return ctx.request.id if n == 1 else f"{ctx.request.id}:{n}"
 
 
+_HELD: ScopeKey[Callable[[], Awaitable[None]]] = ScopeKey("the held rows of the conversation")
+
+
+def holding(ctx: SessionContext, flush: Callable[[], Awaitable[None]]) -> None:
+    """Have ``flush_held`` write, with ``flush``, the rows that the conversation of the attempt's turn holds back."""
+    _HELD.set(ctx, flush)
+
+
+async def flush_held(ctx: SessionContext) -> None:
+    """Write the rows that the conversation of the attempt's turn holds back, if any (see ``_conversation._Ledger``).
+
+    Call it before what another party sees or what outlives the attempt: a
+    publish, a Restate call or send. A held row is lost in a crash, and the
+    retried attempt makes it again; what was seen or called must not
+    depend on it.
+    """
+    flush = _HELD.get(ctx)
+    if flush is not None:
+        await flush()
+
+
 MODEL_ENV = "RESTATE_AGENT_MODEL"
 """The environment variable that names the model agents use when none is given."""