fix(*): keep the context ring right after failed and empty-ending turns

raven/agent/loop/_shared.py

@@ -164,7 +164,11 @@ def _stamp_reasoning_ms(messages: list[dict[str, Any]], response: Any) -> None:
 
 
 # How full a model call left the window, on the assistant entry that stands for
-# it: the reply it produced, or the text a hook put in that reply's place.
+# it: the reply it produced, or the text a hook put in that reply's place. A
+# reply that is not filed leaves it on the newest row the turn keeps instead
+# (``turn_path._stamp_context_used_on_kept``), and a turn that fails or is
+# stopped files its newest reading on the marker it ends on
+# (``_save_broken_turn``).
 # Same underscore-then-rename convention as the spans above; ``_save_turn``
 # writes it as ``context_used``. A resume has no other way to know what the
 # model was sent: the context engine compacts on its own and leaves the stored
@@ -178,8 +182,9 @@ def _window_reading(usage: dict[str, Any] | None) -> int:
     Not what the prompt was billed as. A provider that reports cache reads apart
     from ``prompt_tokens`` (``prompt_tokens_include_cache`` false, the Anthropic
     Messages shape) bills a fraction of a warm prompt while all of it still
-    occupies the window, so the cached counts are added back. Same add-back, for
-    the same reason, as ``transport_failure.never_processed``.
+    occupies the window, so the cached counts are added back -- and only then:
+    where ``prompt_tokens`` already holds them, adding them again would count the
+    warm prompt twice.
     """
     if not usage:
         return 0
@@ -191,13 +196,31 @@ def _window_reading(usage: dict[str, Any] | None) -> int:
     return used
 
 
+def _reads_no_window(response: Any) -> bool:
+    """Whether a call's usage says nothing about how full the window is.
+
+    A response the provider flagged as an upstream transport failure
+    (``upstream_transport_failure``) carries the upstream's account of a call
+    that failed, and that account can be the very evidence that flagged it -- a
+    prompt billed at a size the request cannot have been. The call may never
+    have been read, so its usage is no reading of the window.
+    """
+    classification = getattr(response, "error_classification", None)
+    return getattr(classification, "category", None) == "upstream_transport_failure"
+
+
+def _call_window(response: Any) -> int:
+    """How full a model call left the window, or 0 when it gives no reading."""
+    return 0 if _reads_no_window(response) else _window_reading(getattr(response, "usage", None))
+
+
 def _stamp_context_used(messages: list[dict[str, Any]], response: Any) -> None:
     """Put a model call's window reading on the assistant entry that stands for it.
 
     Absent when the provider reported no usage, and absent means unknown: a
     resume reads the newest entry that has one.
     """
-    used = _window_reading(getattr(response, "usage", None))
+    used = _call_window(response)
     if used > 0 and messages:
         messages[-1][_CONTEXT_USED_KEY] = used
 

raven/agent/loop/turn_path.py

@@ -49,6 +49,7 @@
     Session,
     TurnSynthesisPolicy,
     _appended_by_hook,
+    _call_window,
     _display_label,
     _file_change_payload,
     _file_composed_reply,
@@ -60,7 +61,6 @@
     _stamp_reasoning_ms,
     _stamp_replied_via_tool,
     _strip_inline_images,
-    _window_reading,
     append_hook_note,
     asyncio,
     autofill_resolver,
@@ -358,6 +358,28 @@ def _stamp_turn_observers(messages: list[dict[str, Any]], metadata: dict[str, An
 _TURN_TRANSIENT_KEYS = ("_recovery_synthetic", ATTACHED_IMAGE_KEY)
 
 
+def _stamp_context_used_on_kept(messages: list[dict[str, Any]], response: Any) -> None:
+    """Put the reading of a call whose reply is not filed on the newest row that is.
+
+    An empty reply never becomes a row of its own in the session: recovery
+    stands scaffolding in for it, the loop gives up on it, or it is filed empty
+    and dropped on save. Nor does a provider's error, which is never filed. The
+    live gauge still showed that call's window, and the window it measured ends
+    at the newest row the turn keeps, so the reading goes there. When that row
+    is the turn's question it is already on disk and the stamp never reaches
+    it: the turn either files a row of its own later, which takes a newer
+    reading, or ends on the marker ``_save_broken_turn`` files, which carries
+    it.
+    """
+    used = _call_window(response)
+    if used <= 0:
+        return
+    for message in reversed(messages):
+        if not any(message.get(key) for key in _TURN_TRANSIENT_KEYS):
+            message[_CONTEXT_USED_KEY] = used
+            return
+
+
 # How a turn reports the model saying nothing with no recovery behind it. The
 # texts recovery fails a turn with say what it tried; nothing was tried here,
 # and this says why.
@@ -910,6 +932,7 @@ async def _run_agent_loop(  # noqa: C901 (cc 100: pre-existing, above the ceilin
         turn_log: TurnLog | None = None,
         rungs_used: set[str] | None = None,
         reply_awaited: bool = False,
+        window_reading: dict[str, int] | None = None,
     ) -> tuple[str | None, list[str], list[dict], LoopOutcome]:
         """Run the agent iteration loop.
 
@@ -969,6 +992,11 @@ async def _run_agent_loop(  # noqa: C901 (cc 100: pre-existing, above the ceilin
         thinking-only reply down to, each asked for once. A caller that may run
         the loop again for one turn hands every run the same set; omitted, this
         run keeps a set of its own.
+
+        ``window_reading``, when given, is kept at ``context_used`` with the
+        newest window a call of this loop measured. It outlives the loop the way
+        ``spend`` does: a turn that fails or is stopped files it on its marker,
+        which can be the only row the turn adds.
         """
         spend = spend if spend is not None else TurnSpend(session_key)
         messages = initial_messages
@@ -1599,11 +1627,23 @@ async def _end_on_the_partial(last: LLMResponse) -> str:
                 # keeps it for a resume (``_stamp_context_used``). They differ
                 # in their thresholds, not in what they measure -- and what
                 # they measure is how full the window is, which is not what
-                # the prompt was billed as (see ``_window_reading``).
-                context_used = _window_reading(response.usage)
+                # the prompt was billed as (see ``_window_reading``), and a
+                # call can give none (see ``_call_window``).
+                context_used = _call_window(response)
                 if context_used > 0:
                     window.last_context_used = context_used
-            if usage_sink is not None and response.usage:
+                    if window_reading is not None:
+                        window_reading["context_used"] = context_used
+            if usage_sink is not None and response.usage and context_used <= 0 and usage_sink.get("context_used"):
+                # No reading -- a block of zeros, or an account ``_call_window``
+                # does not take as one -- so the gauge stays on the call before
+                # it, as the reading the turn files does (``window_reading``
+                # above moves on the same terms). The cost is the turn's, a
+                # flow, and moves on regardless. With no reading to keep, the
+                # branch below fills the sink: an answered turn sends it whole.
+                usage_sink["cost_usd"] = spend.cost_usd
+                usage_sink["cost_missing_calls"] = spend.cost_missing_calls
+            elif usage_sink is not None and response.usage:
                 # An explicitly configured window always wins over the live
                 # table -- that is what setting it means. Otherwise the live
                 # window from the model's provider table (e.g. OpenRouter,
@@ -2193,6 +2233,7 @@ async def _end_on_the_partial(last: LLMResponse) -> str:
                     status = "error"
                     error_detail = _llm_failure_detail(clean, verdict)
                     error_full = response_failure_text(clean, response.call_record)
+                    _stamp_context_used_on_kept(messages, response)
                     break
 
                 if recovery.parts:
@@ -2208,6 +2249,9 @@ async def _end_on_the_partial(last: LLMResponse) -> str:
                         recovery.parts.clear()
                         carried_gate = None
 
+                if not (clean or "").strip():
+                    _stamp_context_used_on_kept(messages, response)
+
                 # Empty-response recovery. A reply with no tool call is read for
                 # what it is (``reply_kind``) and ``decide`` names the move; an
                 # empty assistant turn would otherwise end here as a turn with no
@@ -3152,6 +3196,12 @@ async def _tap_episode(_index: int) -> None:
         # turn: a set per attempt would pay for ``high`` and ``low`` again.
         rungs_used: set[str] = set()
 
+        # The newest window a call of this turn measured, held across attempts.
+        # A turn that fails or is stopped after calls that each came back empty
+        # or as an error has filed no row of its own, and the question it would
+        # otherwise end on is already on disk, so its marker carries the reading.
+        reading: dict[str, int] = {}
+
         async def _attempt(seed: list[dict], attempt: int):
             pending.update(rerun=False, reasons=[])
             # The record this attempt persists from, for the rescue paths below as
@@ -3185,6 +3235,7 @@ async def _attempt(seed: list[dict], attempt: int):
                 turn_log=live["log"],
                 rungs_used=rungs_used,
                 reply_awaited=reply_awaited,
+                window_reading=reading,
             )
 
         # Taken BEFORE the first attempt, because the loop appends to the list it is
@@ -3346,6 +3397,7 @@ def _rerun_pending(final_content: str | None, msgs: list[dict], status: str) ->
                 None,
                 streamed,
                 status="cancelled",
+                context_used=reading.get("context_used"),
             )
             raise
         except AnswerlessTurnError as exc:
@@ -3361,6 +3413,7 @@ def _rerun_pending(final_content: str | None, msgs: list[dict], status: str) ->
                 status="failed",
                 reason=str(exc),
                 full_detail=attach_full_detail(exc),
+                context_used=reading.get("context_used"),
             )
             raise
         except Exception as exc:
@@ -3376,6 +3429,7 @@ def _rerun_pending(final_content: str | None, msgs: list[dict], status: str) ->
                 # and a model both read back.
                 reason=bound_failure_text(str(exc)),
                 full_detail=attach_full_detail(exc),
+                context_used=reading.get("context_used"),
             )
             raise
         self._stash_recovery(key, outcome)
@@ -3519,6 +3573,7 @@ def _save_broken_turn(
         reason: str | None = None,
         full_detail: str | None = None,
         inbound_original: str | None = None,
+        context_used: int | None = None,
     ) -> None:
         """Persist what a cancelled or failed turn got as far as producing.
 
@@ -3539,6 +3594,9 @@ def _save_broken_turn(
           ``_marker_failure``).
           ``turn_ended.full_detail`` is the long account beside the bounded
           ``reason``, filed only when there is more to it.
+          ``context_used`` is the newest window the turn measured, filed on the
+          marker because it is the newest row: a resume reads the window from
+          there, and a turn whose calls left no row of their own has no other.
 
         ``received_at`` and ``inbound_original`` describe the turn's question,
         which ``_process_message`` files before the attempt starts and hands
@@ -3603,6 +3661,8 @@ def _save_broken_turn(
                     **({"full_detail": full_detail} if full_detail else {}),
                 },
             }
+            if context_used:
+                marker[_CONTEXT_USED_KEY] = context_used
             tail.append(marker)
             self._save_turn(session, tail, 0, received_at=received_at, inbound_original=inbound_original)
             self.sessions.save(session)

raven/cli/tui_commands.py

@@ -23,7 +23,7 @@
 import sys
 import threading
 from pathlib import Path
-from typing import Callable, Optional, Tuple
+from typing import Any, Callable, Optional, Tuple
 
 import typer
 
@@ -447,6 +447,9 @@ def _agent_loop_factory():
     # turn.send binds a turn's addressee into it and the spine's outlet/sink read
     # it back to tag that turn's events (see build_rpc_spine).
     direct_targets: dict[str, dict[str, str]] = {}
+    # Owned here for the same reason: the spine keeps what each running turn has
+    # measured in it, and turn.cancel reads a stopped turn's reading back.
+    live_usages: dict[str, dict[str, Any]] = {}
     turn_teardown = None
     if agent_loop is not None:
         from raven.core.cron_stack import make_on_cron_job
@@ -462,6 +465,7 @@ def _agent_loop_factory():
             direct_targets=direct_targets,
             readback_texts=cron_readback,
             approval_responder=approval_broker,
+            live_usages=live_usages,
         )
         # Subagent result re-injection submits a SUBAGENT-origin turn.
         agent_loop.subagents.set_submit(turn_scheduler.submit)
@@ -517,6 +521,7 @@ def _agent_loop_factory():
         turn_ids=turn_ids,
         direct_targets=direct_targets,
         build_error=build_error,
+        live_usages=live_usages,
     )
 
     serve_task = asyncio.create_task(server.serve_forever())

raven/rpc/bootstrap.py

@@ -437,6 +437,9 @@ def _agent_loop_factory():
     # turn.send binds a turn's addressee into it and the spine's outlet/sink read
     # it back to tag that turn's events (see build_rpc_spine).
     direct_targets: dict[str, dict[str, str]] = {}
+    # Owned here for the same reason: the spine keeps what each running turn has
+    # measured in it, and turn.cancel reads a stopped turn's reading back.
+    live_usages: dict[str, dict[str, Any]] = {}
     turn_teardown = None
     if agent_loop is not None:
         from raven.core.cron_stack import make_on_cron_job, make_on_session_wake
@@ -459,6 +462,7 @@ def _agent_loop_factory():
             # is not necessarily removing that method.
             approval_responder=approval_responder or approval_broker,
             credential_asker=credential_broker if credential_cards else None,
+            live_usages=live_usages,
         )
         if owns_loop:
             agent_loop.subagents.set_submit(turn_scheduler.submit)
@@ -507,6 +511,7 @@ def _agent_loop_factory():
         send_frame=send_frame,
         default_channel=channel,
         ensure_stack=ensure_stack,
+        live_usages=live_usages,
     )
     if agent_loop is not None:
         _lend_settings_writers(agent_loop, dispatcher)

raven/rpc/methods/__init__.py

@@ -84,6 +84,7 @@ def register_aligned_methods(
     build_error: "RpcError | None" = None,
     send_frame: "Any" = None,
     default_channel: str = "tui",
+    live_usages: "dict[str, dict[str, Any]] | None" = None,
 ) -> None:
     """Register every aligned RPC handler on a dispatcher.
 
@@ -116,6 +117,7 @@ def register_aligned_methods(
         build_error=build_error,
         send_frame=send_frame,
         default_channel=default_channel,
+        live_usages=live_usages,
     )
 
 
@@ -135,6 +137,7 @@ def register_aligned_methods_except_system(
     send_frame: "Any" = None,
     default_channel: str = "tui",
     ensure_stack: "Callable[[], Awaitable[bool]] | None" = None,
+    live_usages: "dict[str, dict[str, Any]] | None" = None,
 ) -> None:
     """Register every aligned RPC handler EXCEPT system.* on a dispatcher.
 
@@ -223,6 +226,7 @@ def register_aligned_methods_except_system(
             # Must equal the channel ``build_rpc_spine`` registered its outlet
             # under; ``turn.py`` documents that a mismatch drops the reply.
             default_channel=default_channel,
+            live_usages=live_usages,
         )
     # confirm.respond — needs a ConfirmBroker to resolve the pending
     # confirm future. Gated like turn.*: when no broker is supplied (demo

raven/rpc/methods/session.py

@@ -684,16 +684,16 @@ async def session_resume(
 def _fill_resumed_context(usage: dict[str, Any], session: Any) -> None:
     """How full the context window is, into a caller's usage mapping.
 
-    Read off the newest stored reply that carries the window its call measured
+    Read off the newest stored row that carries the window a call measured
     (``context_used``): that is the number the meter showed when the turn ended.
     The stored log cannot stand in for it. The context engine compacts on its
     own and leaves the log whole, so ``last_consolidated`` never moves and the
     log is not what the model was sent: a long session's log estimated at forty
     times the window its last call measured, and a short one at a few tokens,
     the system prompt and the tool list being no part of it.
 
-    A session with no reading -- written before replies carried one, or never
-    answered -- falls back to a tiktoken estimate of
+    A session with no reading -- written before rows carried one, or whose
+    calls never reported their usage -- falls back to a tiktoken estimate of
     ``get_history(max_messages=0)``, marked ``context_estimated``. Failures
     leave the zero baseline alone rather than guessing -- a wrong denominator is
     worse than an unknown one.
@@ -707,12 +707,13 @@ def _fill_resumed_context(usage: dict[str, Any], session: Any) -> None:
         usage["context_estimated"] = False
         return
     try:
-        # max_messages=0 is "all of it", which is what both places that build or
-        # measure the real prompt pass (loop/main.py and the consolidator's own
-        # estimator). The default caps at the last 500, and the tail is bounded
-        # in tokens rather than in messages -- consolidation fires on the window
-        # -- so a chatty session on a large-window model sits well past 500 and
-        # the meter would go quiet exactly as it started to matter.
+        # The estimate counts what an engine that leaves compaction to the
+        # consolidator would be handed: max_messages=0 is "all of it", the slice
+        # ``DefaultMemory.candidate_messages`` and the consolidator's own
+        # estimator both take. The default caps at the last 500, and that tail is
+        # bounded in tokens rather than in messages -- consolidation fires on the
+        # window -- so a chatty session on a large-window model sits well past
+        # 500 and the meter would go quiet exactly as it started to matter.
         messages = session.get_history(max_messages=0)
     except Exception:
         logger.exception("session.*: could not read the session history")
@@ -729,11 +730,17 @@ def _fill_resumed_context(usage: dict[str, Any], session: Any) -> None:
     usage["context_estimated"] = True
 
 
+# No context window holds a billion tokens, so a stored reading past this was
+# not written by a call; one past float range would also raise on the way to a
+# percentage. Either is read as no reading.
+_READING_CEILING = 1_000_000_000
+
+
 def _last_window_reading(messages: list[Any]) -> int:
-    """The window the newest reading-carrying reply measured, or 0 for none."""
+    """The window the newest reading-carrying row records, or 0 for none."""
     for message in reversed(messages):
         used = message.get("context_used") if isinstance(message, dict) else None
-        if isinstance(used, int) and not isinstance(used, bool) and used > 0:
+        if isinstance(used, int) and not isinstance(used, bool) and 0 < used <= _READING_CEILING:
             return used
     return 0