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/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 b9c2eeed2..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: @@ -2043,9 +2038,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/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/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/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 diff --git a/tests/acp_stub_server.py b/tests/acp_stub_server.py index 92c66d634..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. @@ -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: 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_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_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_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)) diff --git a/tests/test_subagent_acp.py b/tests/test_subagent_acp.py index 19afa60df..cc77a600e 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,34 @@ 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) + + +@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.""" 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."""