From 1393d3057fa082da323c362cc049f2b57e57173f Mon Sep 17 00:00:00 2001 From: Dizhan Xue Date: Tue, 27 Oct 2026 14:04:46 +0000 Subject: [PATCH 1/5] fix(acp_client): read a Copilot refusal off the reply the caller gets The refusal check read the turn's whole text, every agent message joined. Copilot's refusals are matched by their opening and a short length, so a turn that narrated before the refused call never matched, and the refusal came back as the run's result. The check now reads the last message, the one handed to the caller, here and on the remote pick-up path; the empty-turn check still reads the whole turn. --- raven/acp_client/acp_agent.py | 4 ++-- raven/acp_client/remote_resume.py | 2 +- tests/acp_stub_server.py | 9 ++++++--- tests/test_remote_resume.py | 24 ++++++++++++++++++++++++ tests/test_subagent_github_copilot.py | 18 ++++++++++++++++-- 5 files changed, 49 insertions(+), 8 deletions(-) diff --git a/raven/acp_client/acp_agent.py b/raven/acp_client/acp_agent.py index b9c2eeed2..a6aefa1a6 100644 --- a/raven/acp_client/acp_agent.py +++ b/raven/acp_client/acp_agent.py @@ -2043,9 +2043,9 @@ async def run( f"{detail.strip() or ''}" ) - 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 diff --git a/raven/acp_client/remote_resume.py b/raven/acp_client/remote_resume.py index ab4d6cbc6..1a90e4fc3 100644 --- a/raven/acp_client/remote_resume.py +++ b/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 += ( diff --git a/tests/acp_stub_server.py b/tests/acp_stub_server.py index 92c66d634..57ea3dee5 100644 --- a/tests/acp_stub_server.py +++ b/tests/acp_stub_server.py @@ -90,7 +90,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. @@ -615,9 +615,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": diff --git a/tests/test_remote_resume.py b/tests/test_remote_resume.py index 90c9fe7aa..c8e68ddfc 100644 --- a/tests/test_remote_resume.py +++ b/tests/test_remote_resume.py @@ -151,6 +151,30 @@ async def test_a_picked_up_copilot_reply_that_is_its_providers_refusal_is_an_err assert "for want of credit" in picked.reply +async def test_a_picked_up_refusal_after_an_earlier_message_is_an_error(_pool): + """Read off the reply the pick-up hands back, the last message, as a turn + ended here is: the turn's text joined opens with the narration.""" + link = _Link() + backend = AcpAgentBackend(name="stub@box", command="stub", machine="box", preset="github_copilot") + picking = asyncio.ensure_future(remote_resume.pick_up(_run(), backend, pool=link)) + await asyncio.sleep(0.05) + link.pipe.says(_chunk("I'll read the README.")) + for update in ( + {"sessionUpdate": "tool_call", "toolCallId": "t1", "title": "cat README.md", "status": "pending"}, + {"sessionUpdate": "tool_call_update", "toolCallId": "t1", "status": "completed"}, + ): + link.pipe.says( + {"jsonrpc": "2.0", "method": "session/update", "params": {"sessionId": "sess-1", "update": update}} + ) + link.pipe.says(_chunk("Error: 402 Your credit balance is too low. Please add credits.")) + link.pipe.says({"jsonrpc": "2.0", "id": 3, "result": {"stopReason": "end_turn"}}) + + picked = await asyncio.wait_for(picking, timeout=5) + + assert picked.status == "error" + assert "for want of credit" in picked.reply + + async def test_a_machine_that_cannot_be_asked_or_a_link_that_cannot_attach_keeps_the_record(_pool): link = _Link(tasks=remote_path.NodeCallError("the Raven node on machine 'Lab box' (box) did not answer")) with pytest.raises(remote_resume.UnreachableError, match="did not answer"): diff --git a/tests/test_subagent_github_copilot.py b/tests/test_subagent_github_copilot.py index f739eae4d..b2e27f951 100644 --- a/tests/test_subagent_github_copilot.py +++ b/tests/test_subagent_github_copilot.py @@ -242,7 +242,7 @@ async def _closed_pool(): await close_pool() -def _dispatched(preset: str, reply: str): +def _dispatched(preset: str, reply: str, **env: str): import sys from raven.agent.subagent.backends import build_third_party_backend @@ -254,7 +254,7 @@ def _dispatched(preset: str, reply: str): name="GitHub Copilot", preset=preset, command=f"{command_quote(sys.executable)} {command_quote(str(_STUB))}", - env={"ACP_STUB_REPLY": reply}, + env={"ACP_STUB_REPLY": reply, **env}, ready_timeout_ms=15000, ) ) @@ -279,6 +279,20 @@ async def test_a_dispatch_whose_reply_is_the_refusal_fails(tmp_path: Path, _clos assert _CREDIT in str(caught.value), "the agent's own words travel with the reading" +async def test_a_refusal_after_an_earlier_message_still_fails(tmp_path: Path, _closed_pool) -> None: + """Copilot narrates before a call and can be refused on the next one. The + refusal is read off the reply the caller gets, the last message: the turn's + text joined opens with the narration and never reads as one.""" + from raven.acp_client.acp_agent import AcpProviderRefusalError + + backend = _dispatched("github_copilot", _CREDIT, ACP_STUB_MODE="two_messages") + + with pytest.raises(AcpProviderRefusalError) as caught: + await backend.run("summarize the README", task_id="t1", workspace=tmp_path, executor=None) + + assert caught.value.said == _CREDIT + + async def test_an_answer_that_quotes_such_an_error_is_still_an_answer(tmp_path: Path, _closed_pool) -> None: """A refusal is the whole message: an answer that cites a 500 while debugging something is the work, and failing it would lose the work.""" From cc47de80d492e7c6ec860616a9d7775dd1f60602 Mon Sep 17 00:00:00 2001 From: Dizhan Xue Date: Tue, 27 Oct 2026 14:04:46 +0000 Subject: [PATCH 2/5] fix(rpc): report a wake turn's own usage, not the turn before's A message.complete reports the lane's usage slot. A cron or wake turn never wrote it, and a turn that does not own the lane never clears it, so a wake after a sub-agent result resent that result's usage and cost. An ACP client then added the same cost to the session's running total twice, and every other client was shown the stale numbers. The cron turn now passes a usage sink like any other turn and stashes what it spent. --- raven/rpc/spine.py | 11 +++++++++- tests/test_agent_loop_usage_sink.py | 24 ++++++++++++++++++++++ tests/test_rpc_spine.py | 32 +++++++++++++++++++++++++++++ 3 files changed, 66 insertions(+), 1 deletion(-) diff --git a/raven/rpc/spine.py b/raven/rpc/spine.py index 42e708618..7c9dbefaa 100644 --- a/raven/rpc/spine.py +++ b/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: # 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) diff --git a/tests/test_agent_loop_usage_sink.py b/tests/test_agent_loop_usage_sink.py index da4326f80..5d8da6ecf 100644 --- a/tests/test_agent_loop_usage_sink.py +++ b/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"), diff --git a/tests/test_rpc_spine.py b/tests/test_rpc_spine.py index 5610f1900..7a8929b76 100644 --- a/tests/test_rpc_spine.py +++ b/tests/test_rpc_spine.py @@ -179,6 +179,38 @@ async def test_runner_drives_run_turn_and_stashes_rich_usage(): assert outcome.explicit_reply is True +async def test_a_cron_turn_stashes_its_own_usage_over_the_turn_before(): + """A wake runs on the session's own conversation, and its + ``message.complete`` reports the lane's usage slot. Left unwritten, the slot + still held the turn before -- a sub-agent result that does not own the lane + and so never cleared it -- and that turn's spend was reported a second time.""" + earlier = {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2, "cost_usd": 0.5} + own = {"prompt_tokens": 3, "completion_tokens": 5, "total_tokens": 8, "cost_usd": 0.01, "context_used": 42} + usages: dict[str, dict] = {"tui:c1": dict(earlier)} + runner = RpcTurnRunner(_RunTurnLoop(usage=own, reply_text="done"), FakeEmitter(), usages, {}) + req = TurnRequest(origin=Origin.CRON, source=_src(), text="wake", conversation="tui:c1", turn_id="w1") + _events, emit = _collect() + + await runner.run(req, emit, lambda: []) + + assert usages["tui:c1"] == own + + from raven.rpc.spine import RpcOutlet, _make_rpc_sink + from raven.spine.delivery import DeliveryHub + from raven.spine.events import TurnEnded + + hub, emitter = DeliveryHub(), FakeEmitter() + outlet = RpcOutlet("tui", emitter, {}) + hub.register(outlet) + sink = _make_rpc_sink(hub, outlet, "tui", {}, usages, {}, None) + await sink( + TurnEnded(usage=Usage(0, 0, 0), latency_ms=1.0, explicit_reply=True, conversation_id="tui:c1", turn_id="w1") + ) + + (complete,) = [e for _k, e in emitter.emitted if e["type"] == "message.complete"] + assert complete["payload"]["usage"] == own, "the wake's message.complete reports its own spend" + + async def test_user_turn_receives_tui_approval_capability(tmp_path): executor = _DirectRecordingExecutor() tool = ExecTool(executor=executor, working_dir=str(tmp_path)) From 92abd54f636180c514c1f01eccf2932fc988da8b Mon Sep 17 00:00:00 2001 From: Dizhan Xue Date: Tue, 27 Oct 2026 14:04:46 +0000 Subject: [PATCH 3/5] test(acp_client): pin that a refused or empty turn still books its spend The turn is booked before the empty-turn and refusal checks fail it, and only a comment held that order. The stub's empty prompts now report the usage they were given, so both failures run through a real backend and the ledger is read back. --- tests/acp_stub_server.py | 18 ++++++++++------ tests/test_subagent_acp.py | 43 +++++++++++++++++++++++++++++++++++--- 2 files changed, 52 insertions(+), 9 deletions(-) diff --git a/tests/acp_stub_server.py b/tests/acp_stub_server.py index 57ea3dee5..cbc6d04eb 100644 --- a/tests/acp_stub_server.py +++ b/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. @@ -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 @@ -722,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: diff --git a/tests/test_subagent_acp.py b/tests/test_subagent_acp.py index 19afa60df..c7c5743be 100644 --- a/tests/test_subagent_acp.py +++ b/tests/test_subagent_acp.py @@ -2614,8 +2614,19 @@ async def test_prompt_usage_is_read_only_where_it_counts_the_turn( assert await _counted(tmp_path, **env) == counted -async def _booked(tmp_path: Path, runs: int = 1, *, bound: bool = False, **env: str): - """What the usage ledger holds for ``runs`` turns on one instance of a stub reporting as ``env`` says.""" +async def _booked( + tmp_path: Path, + runs: int = 1, + *, + bound: bool = False, + preset: str | None = None, + raises: type[Exception] | None = None, + **env: str, +): + """What the usage ledger holds for ``runs`` turns on one instance of a stub reporting as ``env`` says. + + ``raises`` is what each turn is expected to fail with. + """ from datetime import datetime, timedelta, timezone from raven.token_wise import usage_context @@ -2632,14 +2643,20 @@ async def _booked(tmp_path: Path, runs: int = 1, *, bound: bool = False, **env: env=dict(cfg.env), snapshot=_snapshot("a", cfg, can_resume=True, can_load=True), registry=registry, + preset=preset, ) since = datetime.now(timezone.utc) - timedelta(seconds=1) ledger = tmp_path / "telemetry" with usage_context.bind("cli:root", {"root_session_key": "cli:root", "telemetry_dir": str(ledger)}): for n in range(runs): - await backend.run( + turn = backend.run( "ping", task_id=f"t{n}", workspace=tmp_path, session_key="cli:root", instance="work", executor=None ) + if raises is None: + await turn + else: + with pytest.raises(raises): + await turn return delegated_usage("cli:root", since, telemetry_dir=ledger) @@ -2689,6 +2706,26 @@ async def test_a_turn_with_neither_counts_nor_cost_books_nothing(tmp_path: Path) assert got.calls == 0 +@pytest.mark.parametrize( + ("preset", "failed"), + [ + (None, {"ACP_STUB_EMPTY_PROMPTS": "1"}), + ("github_copilot", {"ACP_STUB_REPLY": "Error: 402 Your credit balance is too low. Please add credits."}), + ], + ids=["empty", "refused"], +) +async def test_a_turn_that_answered_nothing_usable_still_books_its_spend( + tmp_path: Path, preset: str | None, failed: dict[str, str] +) -> None: + """The turn is booked before the empty-turn and refusal checks fail it: what + it answered is unusable, and what it spent was still spent.""" + from raven.acp_client.acp_agent import AcpEmptyTurnError + + got = await _booked(tmp_path, preset=preset, raises=AcpEmptyTurnError, **_CLAUDE_TURN, **failed) + + assert (got.calls, got.input_tokens, got.output_tokens) == (1, 1200, 34) + + async def test_a_sessions_cumulative_cost_is_booked_once(tmp_path: Path) -> None: """``usage_update.cost`` is the session's cumulative cost, so each turn books what it added: two turns reporting 0.01 then 0.02 cost 0.02, not 0.03.""" From b64743c63201c2139a0237b04f15d6deacb91983 Mon Sep 17 00:00:00 2001 From: Dizhan Xue Date: Tue, 27 Oct 2026 14:04:47 +0000 Subject: [PATCH 4/5] refactor(session): name the turn-marker keys once output_limit_turn_at and budget_stop_turn_at were spelled out by the loop that writes them, the ACP server that reads them and the session that clears them. They are now constants in raven.session.manager; the stored names are unchanged, and the session tests still spell them out. --- raven/acp/methods.py | 3 ++- raven/agent/loop/turn_path.py | 7 ++++--- raven/session/manager.py | 12 ++++++++++-- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/raven/acp/methods.py b/raven/acp/methods.py index b96fadd59..4836f517b 100644 --- a/raven/acp/methods.py +++ b/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 diff --git a/raven/agent/loop/turn_path.py b/raven/agent/loop/turn_path.py index a9172ab83..321819156 100644 --- a/raven/agent/loop/turn_path.py +++ b/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 diff --git a/raven/session/manager.py b/raven/session/manager.py index ceea47263..7ed354ad7 100644 --- a/raven/session/manager.py +++ b/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 From 433cba80158a424ad2286c7a92dfbc454026a7d3 Mon Sep 17 00:00:00 2001 From: Dizhan Xue Date: Tue, 27 Oct 2026 14:04:47 +0000 Subject: [PATCH 5/5] fix(*): read ACP costs and counts by the rules providers/usage.py owns The ACP client's _counted and _turn_cost and the server's _usd re-decided what a reported amount and a token count are. They now use reported_cost and token_count. The difference was the finite check: an inf amount passed all three and a nan passed the client's, and either one as a session's spend baseline left every later turn of that session unpriced. --- raven/acp/updates.py | 14 ++++---------- raven/acp_client/acp_agent.py | 27 +++++++++++---------------- tests/test_acp_updates.py | 4 ++++ tests/test_subagent_acp.py | 8 ++++++++ 4 files changed, 27 insertions(+), 26 deletions(-) diff --git a/raven/acp/updates.py b/raven/acp/updates.py index a9475b825..d4a1cd315 100644 --- a/raven/acp/updates.py +++ b/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: diff --git a/raven/acp_client/acp_agent.py b/raven/acp_client/acp_agent.py index a6aefa1a6..d0677d35d 100644 --- a/raven/acp_client/acp_agent.py +++ b/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: diff --git a/tests/test_acp_updates.py b/tests/test_acp_updates.py index 4f9635251..1e9a565ae 100644 --- a/tests/test_acp_updates.py +++ b/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 diff --git a/tests/test_subagent_acp.py b/tests/test_subagent_acp.py index c7c5743be..cc77a600e 100644 --- a/tests/test_subagent_acp.py +++ b/tests/test_subagent_acp.py @@ -2726,6 +2726,14 @@ async def test_a_turn_that_answered_nothing_usable_still_books_its_spend( assert (got.calls, got.input_tokens, got.output_tokens) == (1, 1200, 34) +@pytest.mark.parametrize("amount", [float("inf"), float("nan"), -0.5, True, "0.5"]) +def test_a_reported_cost_that_is_no_finite_amount_is_unknown_and_keeps_the_baseline(amount: Any) -> None: + """A baseline of ``inf`` or ``nan`` would leave every later turn of the session unpriced.""" + from raven.acp_client.acp_agent import _turn_cost + + assert _turn_cost({"cost": {"amount": amount, "currency": "USD"}}, 0.25) == (None, 0.25) + + async def test_a_sessions_cumulative_cost_is_booked_once(tmp_path: Path) -> None: """``usage_update.cost`` is the session's cumulative cost, so each turn books what it added: two turns reporting 0.01 then 0.02 cost 0.02, not 0.03."""