fix(*): follow up #1121 on Copilot refusals, wake usage and reported costs

raven/acp/methods.py

@@ -44,6 +44,7 @@
 from raven.acp.updates import AcpSession, TurnAlreadyRunningError, UpdateTranslator
 from raven.config import load_config
 from raven.security import redact
+from raven.session.manager import BUDGET_STOP_TURN_AT, OUTPUT_LIMIT_TURN_AT
 
 # Methods in the stable manifest that raven does not serve yet. Answered with
 # method-not-found, which is the same answer an unknown name gets -- the
@@ -967,7 +968,7 @@ def _stop_reason(stop: Any, sessions: Any, session_key: str, filed_before: int)
         if stop != "end_turn" or sessions is None:
             return 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")):
+        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

raven/acp/updates.py

@@ -36,6 +36,7 @@
 
 from raven.acp import protocol
 from raven.acp.tool_kinds import absolute_path, locations, title_for, tool_kind
+from raven.providers.usage import reported_cost
 from raven.security.redact import redact, redact_value
 
 # Every event type ``RpcOutlet``, the spine sink, the DAG bridge and the cron
@@ -389,25 +390,18 @@ def _usage_update(usage: Any, session_cost: float | None = None) -> dict[str, An
     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}
-    if _usd(session_cost) is not None:
-        update["cost"] = {"amount": float(session_cost), "currency": "USD"}
+    if (amount := reported_cost(session_cost)) is not None:
+        update["cost"] = {"amount": amount, "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
+    return reported_cost(usage.get("cost_usd")) if isinstance(usage, dict) else None
 
 
 def _clip_deep(value: Any, depth: int = 0) -> Any:

raven/acp_client/acp_agent.py

@@ -78,6 +78,7 @@
 from raven.home import subagent_role_env
 from raven.mcp.endpoint import McpEndpoints, bridge_command
 from raven.ops import remote_path
+from raven.providers.usage import reported_cost, token_count
 from raven.spine.message import Media
 
 if TYPE_CHECKING:
@@ -177,10 +178,6 @@ def __init__(self, message: str, *, said: str) -> None:
 _RESULT_TEXT_CAP = 2000
 
 
-def _counted(value: Any) -> bool:
-    return isinstance(value, int) and not isinstance(value, bool) and value >= 0
-
-
 def _ledger_counts(counts: Any) -> dict[str, Any] | None:
     """One ACP turn's token counts in the shape the usage ledger reads, or ``None`` without them.
 
@@ -193,19 +190,17 @@ def _ledger_counts(counts: Any) -> dict[str, Any] | None:
     """
     if not isinstance(counts, dict):
         return None
-    read, wrote = counts.get("inputTokens"), counts.get("outputTokens")
-    if not (_counted(read) and _counted(wrote)):
+    read, wrote = token_count(counts.get("inputTokens")), token_count(counts.get("outputTokens"))
+    if read is None or wrote is None:
         return None
-    cached_read, cached_write = counts.get("cachedReadTokens"), counts.get("cachedWriteTokens")
-    thought, total = counts.get("thoughtTokens"), counts.get("totalTokens")
     return {
         "prompt_tokens": read,
         "prompt_tokens_include_cache": False,
         "completion_tokens": wrote,
-        "cache_read_input_tokens": cached_read if _counted(cached_read) else None,
-        "cache_creation_input_tokens": cached_write if _counted(cached_write) else None,
-        "reasoning_tokens": thought if _counted(thought) else 0,
-        "total_tokens": total if _counted(total) else None,
+        "cache_read_input_tokens": token_count(counts.get("cachedReadTokens")),
+        "cache_creation_input_tokens": token_count(counts.get("cachedWriteTokens")),
+        "reasoning_tokens": token_count(counts.get("thoughtTokens")) or 0,
+        "total_tokens": token_count(counts.get("totalTokens")),
     }
 
 
@@ -220,12 +215,12 @@ def _turn_cost(reported: Any, spent_before: float | None) -> tuple[float | None,
     the agent counts some other way, and is read as unknown for the same reason.
     """
     cost = reported.get("cost") if isinstance(reported, dict) else None
-    amount = cost.get("amount") if isinstance(cost, dict) and cost.get("currency") == "USD" else None
-    if not isinstance(amount, (int, float)) or isinstance(amount, bool) or amount < 0:
+    amount = reported_cost(cost.get("amount")) if isinstance(cost, dict) and cost.get("currency") == "USD" else None
+    if amount is None:
         return None, spent_before
     if spent_before is None or amount < spent_before:
-        return None, float(amount)
-    return float(amount) - spent_before, float(amount)
+        return None, amount
+    return amount - spent_before, amount
 
 
 def _holds_session(exc: AcpRemoteError) -> bool:
@@ -2043,9 +2038,9 @@ async def run(
                         f"{detail.strip() or '<no detail>'}"
                     )
 
-                if (refused := self._provider_refusal(text)) is not None:
+                if (refused := self._provider_refusal(collector.reply)) is not None:
                     span.error("the reply is the model provider's refusal")
-                    raise AcpProviderRefusalError(refused, said=text)
+                    raise AcpProviderRefusalError(refused, said=collector.reply)
 
                 if not resumed and self.is_stateful:
                     # Deferred commit, as the cli transport does it: binding a handle

raven/acp_client/remote_resume.py

@@ -123,7 +123,7 @@ async def pick_up(run: RemoteRun, backend: AcpAgentBackend, *, pool: Any = None)
                 f"its frames are on machine {target.label}, task {run.task}",
                 stop_reason,
             )
-        if (refused := backend._provider_refusal(collector.text)) is not None:
+        if (refused := backend._provider_refusal(collector.reply)) is not None:
             return PickedUp("error", refused, stop_reason)
         reply = await backend._finished(collector.reply, stop_reason=stop_reason, span=_Span(), sink=None)
         reply += (

raven/agent/loop/turn_path.py

@@ -139,6 +139,7 @@
 from raven.providers.base import bound_llm_detail, canonical_llm_error, llm_error_summary, parse_llm_error
 from raven.providers.first_byte import first_byte_budget
 from raven.providers.tool_calls import openai_tool_call
+from raven.session.manager import BUDGET_STOP_TURN_AT, OUTPUT_LIMIT_TURN_AT
 from raven.session.refs import TOOL_RESULT_TRUNCATION_MARKER, tool_result_view
 from raven.spine.events import bound_failure_text
 from raven.spine.turn import PROACTIVE_ORIGINS, AnswerlessTurnError
@@ -3301,13 +3302,13 @@ def _rerun_pending(final_content: str | None, msgs: list[dict], status: str) ->
                 # standing -- which is exactly the reading ACP's ``max_tokens``
                 # refinement makes.
                 if turn_hook_meta.get("output_limited"):
-                    session.metadata["output_limit_turn_at"] = _turn_start()
+                    session.metadata[OUTPUT_LIMIT_TURN_AT] = _turn_start()
                 else:
                     # None rather than dropping the key: a save merges its
                     # metadata over the record on disk, so a key left unsaid is
                     # kept rather than cleared (SessionManager._metadata_to_write).
                     # The reader asks whether this is an int, which None is not.
-                    session.metadata["output_limit_turn_at"] = None
+                    session.metadata[OUTPUT_LIMIT_TURN_AT] = None
                 # The same slot for a turn its own budget ended -- the tool cap, the
                 # wall clock or a stalled tool. Its wrap-up reads as an ordinary
                 # answer, so a surface that reports the turn (ACP's stop reason) has
@@ -3316,7 +3317,7 @@ def _rerun_pending(final_content: str | None, msgs: list[dict], status: str) ->
                 # interrupted too, and no budget of its was spent. Written or
                 # cleared on the same terms as above.
                 budget_stopped = outcome.status == "interrupted" and bool(turn_hook_meta.get("budget_stopped"))
-                session.metadata["budget_stop_turn_at"] = _turn_start() if budget_stopped else None
+                session.metadata[BUDGET_STOP_TURN_AT] = _turn_start() if budget_stopped else None
                 # The loop gave up on the model. Its ladder, its rerun and its
                 # salvage have all had their say, so what is left is a turn with
                 # no answer: the failure the handler below files and the lane

raven/rpc/spine.py

@@ -238,10 +238,18 @@ async def _on_review(phase: str, tool_name: str) -> None:
         # delivers a cron.delivered event to every session (the cron:<job_id>
         # conversation has no subscriber, so its hub deliverables no-op). Mirrors
         # the gateway's GatewayTurnRunner read-back path.
+        #
+        # Its usage is stashed as any turn's is. A wake runs on the session's own
+        # conversation, and its message.complete reports the lane's slot: left to
+        # the turn before (one that did not own the lane, so never cleared it),
+        # that turn's spend was reported again.
         if req.origin is Origin.CRON:
             text_sink: dict[str, str] = {}
+            cron_usage: dict[str, Any] = {}
             try:
-                outcome = await self._loop.run_turn(req, emit, drain, stream=False, text_sink=text_sink)
+                outcome = await self._loop.run_turn(
+                    req, emit, drain, stream=False, usage_sink=cron_usage, text_sink=text_sink
+                )
             finally:
                 # A turn that fails stores nothing and the submitter pops by
                 # conversation, so the previous run's text must not be left for
@@ -250,6 +258,7 @@ async def _on_review(phase: str, tool_name: str) -> None:
                     self._readback_texts.pop(req.conversation, None)
             if req.conversation is not None and (text := text_sink.get("text")) is not None:
                 self._readback_texts[req.conversation] = text
+            self._usages[cid] = dict(cron_usage)
             return outcome
         usage_sink: dict[str, Any] = {}
         outcome = await self._loop.run_turn(req, emit, drain, stream=True, usage_sink=usage_sink)

raven/session/manager.py

@@ -36,6 +36,14 @@
 # conversation a writable twin that pickers, sweeps and delivery targets see.
 IMPORTED_CHANNEL = "imported"
 
+# Metadata keys the loop stamps on a session with the message index a cut turn
+# starts at: one stopped at the model's output ceiling, one its own budget ended
+# (the tool cap, the wall clock or a stalled tool). ACP reads them to refine a
+# prompt's ``end_turn``; clearing the session's history forgets them.
+OUTPUT_LIMIT_TURN_AT = "output_limit_turn_at"
+BUDGET_STOP_TURN_AT = "budget_stop_turn_at"
+TURN_MARKERS = (OUTPUT_LIMIT_TURN_AT, BUDGET_STOP_TURN_AT)
+
 
 def _refuse_imported(key: str) -> None:
     if key.partition(":")[0] == IMPORTED_CHANNEL:
@@ -357,14 +365,14 @@ def _forget_turn_markers(self, from_index: int) -> None:
         """Clear the turn markers whose turn starts at ``from_index`` or later.
 
         The loop stamps a cut turn with the index its rows start at
-        (``output_limit_turn_at``, ``budget_stop_turn_at``) and ACP reports a
+        (:data:`TURN_MARKERS`) and ACP reports a
         prompt as cut when one sits at or past where that prompt began. Once the
         messages from ``from_index`` are gone, the next prompt begins there, and
         a command that writes no marker of its own would be reported cut by a
         turn that no longer exists. Cleared as None rather than dropped, per the
         removal rule in ``SessionManager._metadata_to_write``.
         """
-        for key in ("output_limit_turn_at", "budget_stop_turn_at"):
+        for key in TURN_MARKERS:
             at = self.metadata.get(key)
             if isinstance(at, int) and at >= from_index:
                 self.metadata[key] = None

tests/acp_stub_server.py

@@ -11,7 +11,8 @@
                      ``ACP_STUB_REPLY`` replaces the chunk's text (``pong``),
                      ``ACP_STUB_USAGE_UPDATE`` the ``usage_update`` body (JSON), and
                      ``ACP_STUB_PROMPT_USAGE`` adds a ``usage`` to the prompt response (JSON).
-                     ``ACP_STUB_EMPTY_PROMPTS=N`` ends the first N prompts empty, in any mode.
+                     ``ACP_STUB_EMPTY_PROMPTS=N`` ends the first N prompts empty, in any mode,
+                     still with the ``ACP_STUB_PROMPT_USAGE`` they spent.
                      ``ACP_STUB_AGENT_NAME`` replaces ``agentInfo.name`` (``stub-agent``).
                      ``ACP_STUB_AGENT_VERSION`` replaces ``agentInfo.version`` (``9.9.9``); empty drops it.
                      ``ACP_STUB_COST_PER_PROMPT`` reports a cumulative cost growing by it each prompt.
@@ -90,7 +91,7 @@
                      loop, so this is what fails if answering one blocks it.
 - ``two_messages`` - a preamble, the tool calls it announced, then the answer, in the
                      order measured on codex-acp. One turn, two messages, and no
-                     boundary in the chunks themselves.
+                     boundary in the chunks themselves. ``ACP_STUB_REPLY`` replaces the answer.
 - ``cancelled``    - streams one chunk, then reports ``stopReason: "cancelled"``. The
                      shape measured on codex-acp when a turn is torn down part-way: a
                      reply that reads finished and is not.
@@ -338,13 +339,21 @@ def _permission_input() -> dict:
 _PROMPTS = 0
 
 
+def _ended() -> dict:
+    """A finished prompt's result, with ``ACP_STUB_PROMPT_USAGE`` as its ``usage`` when set."""
+    answered: dict = {"stopReason": "end_turn"}
+    if counts := os.environ.get("ACP_STUB_PROMPT_USAGE"):
+        answered["usage"] = json.loads(counts)
+    return answered
+
+
 def handle_prompt(request_id, params) -> None:
     global _PROMPTS
     _PROMPTS += 1
     session_id = params.get("sessionId") or "stub-session-1"
     if _PROMPTS <= int(os.environ.get("ACP_STUB_EMPTY_PROMPTS", "0")):
         # The first N prompts end with nothing said, whatever the mode.
-        ok(request_id, {"stopReason": "end_turn"})
+        ok(request_id, _ended())
         return
     if MODE == "dies_mid_turn":
         # Half an answer, then the process is gone: the prompt went out and
@@ -615,9 +624,12 @@ def handle_prompt(request_id, params) -> None:
         update(session_id, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "look."}})
         update(session_id, {"sessionUpdate": "tool_call", "toolCallId": "t1", "title": "ls", "status": "pending"})
         update(session_id, {"sessionUpdate": "tool_call_update", "toolCallId": "t1", "status": "completed"})
-        update(session_id, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "it is "}})
+        answer = os.environ.get("ACP_STUB_REPLY")
+        opening, rest = ("it is ", "a repo.") if answer is None else ("", answer)
+        if opening:
+            update(session_id, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": opening}})
         update(session_id, {"sessionUpdate": "usage_update", "size": 1000, "used": 7})
-        update(session_id, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "a repo."}})
+        update(session_id, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": rest}})
         ok(request_id, {"stopReason": "end_turn"})
         return
     if MODE == "ends_on_a_step":
@@ -719,10 +731,7 @@ def handle_prompt(request_id, params) -> None:
         # Cumulative, as the schema defines it: the process's whole spend so far.
         reported["cost"] = {"amount": round(float(per_prompt) * _PROMPTS, 6), "currency": "USD"}
     update(session_id, {"sessionUpdate": "usage_update", **reported})
-    answered: dict = {"stopReason": "end_turn"}
-    if counts := os.environ.get("ACP_STUB_PROMPT_USAGE"):
-        answered["usage"] = json.loads(counts)
-    ok(request_id, answered)
+    ok(request_id, _ended())
 
 
 def handle_cancel(params) -> None:

tests/test_acp_updates.py

@@ -488,6 +488,10 @@ def test_an_unpriced_turn_leaves_the_total_where_it_was(self):
         what was reported stands, and before any price there is none to send."""
         assert self._costs(_session(), None, 0.01, None) == pytest.approx([None, 0.01, 0.01])
 
+    def test_a_cost_that_is_no_finite_amount_leaves_the_total_where_it_was(self):
+        """``inf`` added once would be the session's total for good."""
+        assert self._costs(_session(), 0.01, float("inf"), 0.02) == pytest.approx([0.01, 0.01, 0.03])
+
     def test_a_session_that_was_resumed_counts_on_from_its_earlier_spend(self):
         session = _session()
         session.cost_usd = 1.5

tests/test_agent_loop_usage_sink.py

@@ -138,6 +138,30 @@ async def test_usage_sink_carries_context_gauge_and_cost(workspace):
     assert sink["cost_missing_calls"] == 1
 
 
+@pytest.mark.asyncio
+async def test_a_turn_run_without_a_stream_fills_the_usage_sink_too(workspace):
+    """A cron or wake turn runs without a stream; its ``message.complete``
+    reports what it spent from the same sink a streamed turn fills."""
+    provider = UsageProvider("stub", prompt_tokens=6000, completion_tokens=2000)
+    agent = _make_agent(workspace, provider, model="stub", window=40000)
+    sink: dict = {}
+    source = Source(channel="test", chat_id="c1", sender_id="user", chat_type=ChatType.DM)
+
+    async def emit(_event) -> None:
+        return None
+
+    await agent.run_turn(
+        TurnRequest(origin=Origin.CRON, source=source, text="wake", conversation="test:c1"),
+        emit,
+        lambda: [],
+        stream=False,
+        usage_sink=sink,
+        text_sink={},
+    )
+
+    assert (sink["prompt_tokens"], sink["completion_tokens"], sink["context_used"]) == (6000, 2000, 8000)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("extra", "used", "percent"),