fix(*): report a delegated turn's real outcome and cost

docs/specs/2026-08-21-acp-agent-compatibility.md

@@ -74,11 +74,11 @@ build is validated against it.
 | `tool_call_update` status | **Supported** | `completed` or `failed`, from the emit site's verdict: the loop derives it from the registry's own failure convention (the model-facing text of a failed tool starts with `Error`), which is the same rule the retry-hint logic uses. A tool that reports a friendly `Error: ...` message is therefore classified as failed here too -- breaking the convention mislabels the transcript row as well. |
 | `ToolCallContent` `diff` | **Supported** | Structured `{path, newText, oldText?}` carried from the write tools, which hold both versions of the file. `oldText` is omitted only for a file that did not exist -- the schema defines its absence as "new file", so omitting it for a file whose previous content is merely unavailable would render every line of a rewrite as an addition. |
 | `ToolCallContent` `terminal` | **Not supported** | See `terminal/*` below. |
-| `stopReason` | `end_turn`, `cancelled` | `max_tokens` and `max_turn_requests` have no source in Raven and are never sent. |
+| `stopReason` | `end_turn`, `cancelled`, `max_turn_requests`, `max_tokens` | A turn that would otherwise end `end_turn` is reported `max_turn_requests` when the agent loop's own budget ended it (the tool-call cap, the wall clock or a stalled tool) and a wrap-up answered in its place, and `max_tokens` when generation stopped at the model's output ceiling. The budget is read first. Either way the reply reads complete and is not, and the stop reason is the one place a delegating client can learn that. |
 | `refusal` as a stop reason | **Not sent** | It would come from the runtime's `action_blocked` notice, and that notice carries no turn id: `_run_turn` has none to give it, and the runtime shares a lane with the client's turn (see `_owns_lane`), so a refusal seen on this session cannot be shown to belong to this prompt. Latching it anyway made a *foreign* turn's block the reported outcome of a turn that completed normally, which is a false statement about this turn; reporting `end_turn` is merely a less specific true one. The refusal itself is not lost -- it is delivered as message content, which is what the person reads. Restoring it needs a turn id on the notice, which means threading one through the agent loop. |
 | A failed turn | **Explained, then ended** | A prompt is never answered with a JSON-RPC error. Measured from the other direction on codex-acp: an error in reply to a turn-shaped request makes clients tear down the whole turn. The failure is surfaced as message content and the turn ends with a stop reason. |
 | `plan` | **Not sent** | `PlanEntry.priority` is required (`high`/`medium`/`low`) and Raven's DAG has no source for it. A plan would have to be invented. |
-| `usage_update` | **Supported** | `used` and `size` are the context-window numbers, `cost` is the estimated dollar figure. Sent on the turn's completion, before the prompt is answered, so the client has it while the turn still exists. Withheld entirely when the window numbers are not real: a `size` of zero has a client drawing a full bar or dividing by it, and an update of zeroes is not the same statement as no update. A cost of zero, by contrast, is reported -- a cached reply cost nothing, which is different from not knowing. |
+| `usage_update` | **Supported** | `used` and `size` are the context-window numbers. `cost` is the session's cumulative cost, as the schema defines it: what the session had spent when it was opened on this connection (read from its usage record, which outlives the process), plus what each turn since reported, delegations included. It sums only prices providers reported, and is omitted when none did or when the earlier spend could not be read. Sent on the turn's completion, before the prompt is answered, so the client has it while the turn still exists. Withheld entirely when the window numbers are not real: a `size` of zero has a client drawing a full bar or dividing by it, and an update of zeroes is not the same statement as no update. A cost of zero, by contrast, is reported -- a cached reply cost nothing, which is different from not knowing. |
 | `session_info_update` | **Not sent yet** | The title is available (it is what `session/list` reports), but nothing changes it mid-session today. |
 | `available_commands_update` | **Supported** | A deliberate white-list of six user-facing actions (`deep-research`, `playbook`, `cron`, `sentinel`, `sessions`, `tracing`), announced once per session on `session/new`, `session/load` and `session/resume`. Not a reflection of the Typer app: the CLI surface includes verbs whose meaning is internal to a terminal, and reflecting it would put those in a command panel and drift with every new verb. No per-turn updates -- the list is static for the life of a connection. |
 | `current_mode_update` / `session/set_mode` | **Not supported** | Raven has no mode concept. Declaring a mode with no state behind it is worse than not declaring one. |

raven/acp/methods.py

@@ -379,12 +379,11 @@ async def _session_load(self, params: dict[str, Any]) -> dict[str, Any]:
         self._bind_workdir(session_id, cwd)
         session = self._translator.get(session_id)
         if session is None:
-            session = AcpSession(
-                session_id=session_id,
-                session_key=session_id,
-                cwd=cwd,
-                subscription_id=await self._subscribe(session_id),
-            )
+            session = AcpSession(session_id=session_id, session_key=session_id, cwd=cwd)
+            # Before the subscription, so nothing is awaited between subscribing
+            # and adding: an event in that gap finds no session and is dropped.
+            await self._seed_cost(session)
+            session.subscription_id = await self._subscribe(session_id)
             self._translator.add(session)
         else:
             # Loading a session this connection already holds. Its stream is
@@ -400,6 +399,26 @@ async def _session_load(self, params: dict[str, Any]) -> dict[str, Any]:
         self._announce_commands(session_id)
         return self._with_modes({}, session_id)
 
+    async def _seed_cost(self, session: AcpSession) -> None:
+        """Start a session opened here from what it had already spent.
+
+        ``usage_update.cost`` is the session's cumulative cost, and turns run
+        before this process -- or on another connection -- are part of it. Read
+        from ``session.usage``, the account that outlives the process, once per
+        session opened rather than per turn: it walks every day of telemetry
+        since the session began. Unreadable, the session sends no cost at all; a
+        total counted from here would understate it.
+        """
+        try:
+            spent = await self._call("session.usage", {"session_id": session.session_key})
+        except Exception as exc:  # noqa: BLE001 - accounting must not refuse the session
+            logger.warning("acp: what session {} had spent could not be read: {}", session.session_id, exc)
+            session.cost_known = False
+            return
+        cost = spent.get("cost_usd") if isinstance(spent, dict) else None
+        if isinstance(cost, (int, float)) and not isinstance(cost, bool) and cost >= 0:
+            session.cost_usd = float(cost)
+
     async def _session_resume(self, params: dict[str, Any]) -> dict[str, Any]:
         """Reopen a stored session for its own state, without replaying it.
 
@@ -435,12 +454,11 @@ async def _session_resume(self, params: dict[str, Any]) -> dict[str, Any]:
         self._bind_workdir(session_id, cwd)
         session = self._translator.get(session_id)
         if session is None:
-            session = AcpSession(
-                session_id=session_id,
-                session_key=session_id,
-                cwd=cwd,
-                subscription_id=await self._subscribe(session_id),
-            )
+            session = AcpSession(session_id=session_id, session_key=session_id, cwd=cwd)
+            # Before the subscription, so nothing is awaited between subscribing
+            # and adding: an event in that gap finds no session and is dropped.
+            await self._seed_cost(session)
+            session.subscription_id = await self._subscribe(session_id)
             self._translator.add(session)
         else:
             session.cwd = cwd
@@ -922,9 +940,15 @@ def _turn_meta(sessions: Any, session_key: str, filed_before: int) -> dict[str,
 
     @staticmethod
     def _stop_reason(stop: Any, sessions: Any, session_key: str, filed_before: int) -> Any:
-        """``stop``, refined to ``max_tokens`` when this turn hit the ceiling.
+        """``stop``, refined to the limit that cut this turn short, if one did.
+
+        ``max_turn_requests`` when the turn's own budget ended it (the tool cap,
+        the wall clock or a stalled tool) and a wrap-up answered in its place;
+        ``max_tokens`` when generation stopped at the output ceiling. The budget
+        is read first: it is what ended the turn, and a wrap-up that was then
+        cut too does not make the turn any less over budget.
 
-        The protocol already has a word for it, and the schema's own enum
+        The protocol already has a word for each, and the schema's own enum
         carries it -- so a turn cut at the output ceiling is reported the way
         every conforming agent reports one, rather than through a private
         ``_meta`` key the client would have to be taught and that no other agent
@@ -942,8 +966,12 @@ def _stop_reason(stop: Any, sessions: Any, session_key: str, filed_before: int)
         """
         if stop != "end_turn" or sessions is None:
             return stop
-        at = getattr(sessions.get_or_create(session_key), "metadata", {}).get("output_limit_turn_at")
-        return "max_tokens" if isinstance(at, int) and at >= filed_before else stop
+        metadata = getattr(sessions.get_or_create(session_key), "metadata", {})
+        for key, refined in (("budget_stop_turn_at", "max_turn_requests"), ("output_limit_turn_at", "max_tokens")):
+            at = metadata.get(key)
+            if isinstance(at, int) and at >= filed_before:
+                return refined
+        return stop
 
     async def _session_cancel(self, params: dict[str, Any]) -> None:
         """Cancel the session's turn, and make sure its prompt is answered.

raven/acp/protocol.py

@@ -89,14 +89,12 @@
 )
 """Every value ``PromptResponse.stopReason`` may take.
 
-``max_turn_requests`` has no source in raven and is therefore never sent: it
-describes a limit the agent loop does not report hitting. ``max_tokens`` is sent
-for a turn whose generation stopped at the model's output ceiling, which is what
-a delegating client needs in order to weigh the reply -- refined from
-``end_turn`` in `raven/acp/methods.py`, and never over a reason that already
-says why the turn ended. Both stay in the set because it is the *schema's* enum,
-and a translator that latched one of them would be caught by the schema check
-rather than by this constant.
+``max_turn_requests`` is sent for a turn the agent loop's own budget ended (the
+tool cap, the wall clock or a stalled tool) and ``max_tokens`` for one whose
+generation stopped at the model's output ceiling: either way the reply is a
+wrap-up rather than the work, which is what a delegating client needs in order
+to weigh it. Both are refined from ``end_turn`` in `raven/acp/methods.py`, and
+never over a reason that already says why the turn ended.
 """
 
 

raven/acp/updates.py

@@ -197,7 +197,7 @@ def _meta_for(payload: dict[str, Any]) -> dict[str, Any] | None:
     return {"raven.target": target}
 
 
-def translate(event: Any, *, cwd: str | None = None) -> Translated:
+def translate(event: Any, *, cwd: str | None = None, session_cost: float | None = None) -> Translated:
     """Read one wire event. Pure: no state, no I/O, no event loop.
 
     Kept pure so the frames it produces can be validated against the official
@@ -207,6 +207,9 @@ def translate(event: Any, *, cwd: str | None = None) -> Translated:
     An event this has no branch for yields an empty result rather than raising.
     A translator that crashed on an unrecognised event would take the whole
     connection down over a wire event somebody added for the web client.
+
+    ``session_cost`` is the session's cumulative cost, which the caller keeps:
+    it is what ACP's ``usage_update.cost`` reports, and no one event carries it.
     """
     if not isinstance(event, dict):
         return Translated()
@@ -242,7 +245,7 @@ def translate(event: Any, *, cwd: str | None = None) -> Translated:
         # latched one earlier; ``end_turn`` here is the default that a plain
         # completion means. The usage rides out first, so the client has the
         # turn's cost before the turn is over rather than after.
-        usage = _usage_update(payload.get("usage"))
+        usage = _usage_update(payload.get("usage"), session_cost)
         return Translated(updates=() if usage is None else (usage,), stop="end_turn")
 
     if kind == "media":
@@ -361,8 +364,8 @@ def _file_uri(path: str) -> str:
         return "file://" + quote(path, safe="/")
 
 
-def _usage_update(usage: Any) -> dict[str, Any] | None:
-    """How full the context window is, and what the turn cost.
+def _usage_update(usage: Any, session_cost: float | None = None) -> dict[str, Any] | None:
+    """How full the context window is, and what the session has cost so far.
 
     The one place the rich accounting already on the internal wire maps cleanly
     onto ACP: ``used`` and ``size`` are the context numbers, and ``Cost`` wants
@@ -373,8 +376,11 @@ def _usage_update(usage: Any) -> dict[str, Any] | None:
     no usage at all (a cached reply, a hook short-circuit) has nothing to say
     here -- an update of zeroes is not the same statement as no update.
 
-    The currency is USD because cost_usd carries only verified dollar amounts
-    reported by the provider, never a locally estimated price.
+    ``cost`` is ``session_cost``, not the turn's ``cost_usd``: ACP defines it as
+    the session's cumulative cost, and the turn's alone made a client show the
+    last turn's spend as the whole session's. The currency is USD because the
+    total sums only dollar amounts providers reported, never a locally estimated
+    price.
     """
     if not isinstance(usage, dict):
         return None
@@ -383,12 +389,27 @@ def _usage_update(usage: Any) -> dict[str, Any] | None:
     if not isinstance(used, int) or not isinstance(size, int) or size <= 0 or used < 0:
         return None
     update: dict[str, Any] = {"sessionUpdate": "usage_update", "used": used, "size": size}
-    cost = usage.get("cost_usd")
-    if isinstance(cost, (int, float)) and not isinstance(cost, bool) and cost >= 0:
-        update["cost"] = {"amount": float(cost), "currency": "USD"}
+    if _usd(session_cost) is not None:
+        update["cost"] = {"amount": float(session_cost), "currency": "USD"}
     return update
 
 
+def _usd(value: Any) -> float | None:
+    """``value`` as a reported dollar amount, or None when it is not one."""
+    if isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0:
+        return float(value)
+    return None
+
+
+def _turn_spend(event: Any) -> float | None:
+    """What the turn a ``message.complete`` ends reported spending, or None for any other event."""
+    if not isinstance(event, dict) or event.get("type") != "message.complete":
+        return None
+    payload = event.get("payload")
+    usage = payload.get("usage") if isinstance(payload, dict) else None
+    return _usd(usage.get("cost_usd")) if isinstance(usage, dict) else None
+
+
 def _clip_deep(value: Any, depth: int = 0) -> Any:
     """Clip every string in a JSON-shaped value, structure intact.
 
@@ -698,6 +719,13 @@ class AcpSession:
     cwd: str
     subscription_id: str | None = None
     turn: _Turn | None = None
+    cost_usd: float | None = None
+    """What the session has cost so far, in provider-reported USD: what it had
+    spent when it was opened here, then each turn's spend added as it ends. None
+    until some call reports a price."""
+    cost_known: bool = True
+    """False when what the session had spent before it was opened here could not
+    be read: a total counted from zero would understate it, so none is sent."""
 
 
 class UpdateTranslator:
@@ -919,7 +947,9 @@ async def send_frame(self, frame: Any) -> None:
         await self._deliver(session, params.get("event"))
 
     async def _deliver(self, session: AcpSession, event: Any) -> None:
-        result = translate(event, cwd=session.cwd)
+        if (spent := _turn_spend(event)) is not None:
+            session.cost_usd = (session.cost_usd or 0.0) + spent
+        result = translate(event, cwd=session.cwd, session_cost=session.cost_usd if session.cost_known else None)
         # Held thoughts go out before anything that has to follow them: a
         # chunk of the answer, a tool call, an ending. An event that carries
         # none of those (a thought, or nothing at all) leaves them held, which