diff --git a/docs/specs/2026-08-21-acp-agent-compatibility.md b/docs/specs/2026-08-21-acp-agent-compatibility.md index 176e3f420..ec71fbe20 100644 --- a/docs/specs/2026-08-21-acp-agent-compatibility.md +++ b/docs/specs/2026-08-21-acp-agent-compatibility.md @@ -74,11 +74,11 @@ build is validated against it. | `tool_call_update` status | **Supported** | `completed` or `failed`, from the emit site's verdict: the loop derives it from the registry's own failure convention (the model-facing text of a failed tool starts with `Error`), which is the same rule the retry-hint logic uses. A tool that reports a friendly `Error: ...` message is therefore classified as failed here too -- breaking the convention mislabels the transcript row as well. | | `ToolCallContent` `diff` | **Supported** | Structured `{path, newText, oldText?}` carried from the write tools, which hold both versions of the file. `oldText` is omitted only for a file that did not exist -- the schema defines its absence as "new file", so omitting it for a file whose previous content is merely unavailable would render every line of a rewrite as an addition. | | `ToolCallContent` `terminal` | **Not supported** | See `terminal/*` below. | -| `stopReason` | `end_turn`, `cancelled` | `max_tokens` and `max_turn_requests` have no source in Raven and are never sent. | +| `stopReason` | `end_turn`, `cancelled`, `max_turn_requests`, `max_tokens` | A turn that would otherwise end `end_turn` is reported `max_turn_requests` when the agent loop's own budget ended it (the tool-call cap, the wall clock or a stalled tool) and a wrap-up answered in its place, and `max_tokens` when generation stopped at the model's output ceiling. The budget is read first. Either way the reply reads complete and is not, and the stop reason is the one place a delegating client can learn that. | | `refusal` as a stop reason | **Not sent** | It would come from the runtime's `action_blocked` notice, and that notice carries no turn id: `_run_turn` has none to give it, and the runtime shares a lane with the client's turn (see `_owns_lane`), so a refusal seen on this session cannot be shown to belong to this prompt. Latching it anyway made a *foreign* turn's block the reported outcome of a turn that completed normally, which is a false statement about this turn; reporting `end_turn` is merely a less specific true one. The refusal itself is not lost -- it is delivered as message content, which is what the person reads. Restoring it needs a turn id on the notice, which means threading one through the agent loop. | | A failed turn | **Explained, then ended** | A prompt is never answered with a JSON-RPC error. Measured from the other direction on codex-acp: an error in reply to a turn-shaped request makes clients tear down the whole turn. The failure is surfaced as message content and the turn ends with a stop reason. | | `plan` | **Not sent** | `PlanEntry.priority` is required (`high`/`medium`/`low`) and Raven's DAG has no source for it. A plan would have to be invented. | -| `usage_update` | **Supported** | `used` and `size` are the context-window numbers, `cost` is the estimated dollar figure. Sent on the turn's completion, before the prompt is answered, so the client has it while the turn still exists. Withheld entirely when the window numbers are not real: a `size` of zero has a client drawing a full bar or dividing by it, and an update of zeroes is not the same statement as no update. A cost of zero, by contrast, is reported -- a cached reply cost nothing, which is different from not knowing. | +| `usage_update` | **Supported** | `used` and `size` are the context-window numbers. `cost` is the session's cumulative cost, as the schema defines it: what the session had spent when it was opened on this connection (read from its usage record, which outlives the process), plus what each turn since reported, delegations included. It sums only prices providers reported, and is omitted when none did or when the earlier spend could not be read. Sent on the turn's completion, before the prompt is answered, so the client has it while the turn still exists. Withheld entirely when the window numbers are not real: a `size` of zero has a client drawing a full bar or dividing by it, and an update of zeroes is not the same statement as no update. A cost of zero, by contrast, is reported -- a cached reply cost nothing, which is different from not knowing. | | `session_info_update` | **Not sent yet** | The title is available (it is what `session/list` reports), but nothing changes it mid-session today. | | `available_commands_update` | **Supported** | A deliberate white-list of six user-facing actions (`deep-research`, `playbook`, `cron`, `sentinel`, `sessions`, `tracing`), announced once per session on `session/new`, `session/load` and `session/resume`. Not a reflection of the Typer app: the CLI surface includes verbs whose meaning is internal to a terminal, and reflecting it would put those in a command panel and drift with every new verb. No per-turn updates -- the list is static for the life of a connection. | | `current_mode_update` / `session/set_mode` | **Not supported** | Raven has no mode concept. Declaring a mode with no state behind it is worse than not declaring one. | diff --git a/raven/acp/methods.py b/raven/acp/methods.py index 89cf8e8a6..b96fadd59 100644 --- a/raven/acp/methods.py +++ b/raven/acp/methods.py @@ -379,12 +379,11 @@ async def _session_load(self, params: dict[str, Any]) -> dict[str, Any]: self._bind_workdir(session_id, cwd) session = self._translator.get(session_id) if session is None: - session = AcpSession( - session_id=session_id, - session_key=session_id, - cwd=cwd, - subscription_id=await self._subscribe(session_id), - ) + session = AcpSession(session_id=session_id, session_key=session_id, cwd=cwd) + # Before the subscription, so nothing is awaited between subscribing + # and adding: an event in that gap finds no session and is dropped. + await self._seed_cost(session) + session.subscription_id = await self._subscribe(session_id) self._translator.add(session) else: # Loading a session this connection already holds. Its stream is @@ -400,6 +399,26 @@ async def _session_load(self, params: dict[str, Any]) -> dict[str, Any]: self._announce_commands(session_id) return self._with_modes({}, session_id) + async def _seed_cost(self, session: AcpSession) -> None: + """Start a session opened here from what it had already spent. + + ``usage_update.cost`` is the session's cumulative cost, and turns run + before this process -- or on another connection -- are part of it. Read + from ``session.usage``, the account that outlives the process, once per + session opened rather than per turn: it walks every day of telemetry + since the session began. Unreadable, the session sends no cost at all; a + total counted from here would understate it. + """ + try: + spent = await self._call("session.usage", {"session_id": session.session_key}) + except Exception as exc: # noqa: BLE001 - accounting must not refuse the session + logger.warning("acp: what session {} had spent could not be read: {}", session.session_id, exc) + session.cost_known = False + return + cost = spent.get("cost_usd") if isinstance(spent, dict) else None + if isinstance(cost, (int, float)) and not isinstance(cost, bool) and cost >= 0: + session.cost_usd = float(cost) + async def _session_resume(self, params: dict[str, Any]) -> dict[str, Any]: """Reopen a stored session for its own state, without replaying it. @@ -435,12 +454,11 @@ async def _session_resume(self, params: dict[str, Any]) -> dict[str, Any]: self._bind_workdir(session_id, cwd) session = self._translator.get(session_id) if session is None: - session = AcpSession( - session_id=session_id, - session_key=session_id, - cwd=cwd, - subscription_id=await self._subscribe(session_id), - ) + session = AcpSession(session_id=session_id, session_key=session_id, cwd=cwd) + # Before the subscription, so nothing is awaited between subscribing + # and adding: an event in that gap finds no session and is dropped. + await self._seed_cost(session) + session.subscription_id = await self._subscribe(session_id) self._translator.add(session) else: session.cwd = cwd @@ -922,9 +940,15 @@ def _turn_meta(sessions: Any, session_key: str, filed_before: int) -> dict[str, @staticmethod def _stop_reason(stop: Any, sessions: Any, session_key: str, filed_before: int) -> Any: - """``stop``, refined to ``max_tokens`` when this turn hit the ceiling. + """``stop``, refined to the limit that cut this turn short, if one did. + + ``max_turn_requests`` when the turn's own budget ended it (the tool cap, + the wall clock or a stalled tool) and a wrap-up answered in its place; + ``max_tokens`` when generation stopped at the output ceiling. The budget + is read first: it is what ended the turn, and a wrap-up that was then + cut too does not make the turn any less over budget. - The protocol already has a word for it, and the schema's own enum + The protocol already has a word for each, and the schema's own enum carries it -- so a turn cut at the output ceiling is reported the way every conforming agent reports one, rather than through a private ``_meta`` key the client would have to be taught and that no other agent @@ -942,8 +966,12 @@ def _stop_reason(stop: Any, sessions: Any, session_key: str, filed_before: int) """ if stop != "end_turn" or sessions is None: return stop - at = getattr(sessions.get_or_create(session_key), "metadata", {}).get("output_limit_turn_at") - return "max_tokens" if isinstance(at, int) and at >= filed_before else stop + metadata = getattr(sessions.get_or_create(session_key), "metadata", {}) + for key, refined in (("budget_stop_turn_at", "max_turn_requests"), ("output_limit_turn_at", "max_tokens")): + at = metadata.get(key) + if isinstance(at, int) and at >= filed_before: + return refined + return stop async def _session_cancel(self, params: dict[str, Any]) -> None: """Cancel the session's turn, and make sure its prompt is answered. diff --git a/raven/acp/protocol.py b/raven/acp/protocol.py index cf02e42c7..887f36f15 100644 --- a/raven/acp/protocol.py +++ b/raven/acp/protocol.py @@ -89,14 +89,12 @@ ) """Every value ``PromptResponse.stopReason`` may take. -``max_turn_requests`` has no source in raven and is therefore never sent: it -describes a limit the agent loop does not report hitting. ``max_tokens`` is sent -for a turn whose generation stopped at the model's output ceiling, which is what -a delegating client needs in order to weigh the reply -- refined from -``end_turn`` in `raven/acp/methods.py`, and never over a reason that already -says why the turn ended. Both stay in the set because it is the *schema's* enum, -and a translator that latched one of them would be caught by the schema check -rather than by this constant. +``max_turn_requests`` is sent for a turn the agent loop's own budget ended (the +tool cap, the wall clock or a stalled tool) and ``max_tokens`` for one whose +generation stopped at the model's output ceiling: either way the reply is a +wrap-up rather than the work, which is what a delegating client needs in order +to weigh it. Both are refined from ``end_turn`` in `raven/acp/methods.py`, and +never over a reason that already says why the turn ended. """ diff --git a/raven/acp/updates.py b/raven/acp/updates.py index 7850445b2..a9475b825 100644 --- a/raven/acp/updates.py +++ b/raven/acp/updates.py @@ -197,7 +197,7 @@ def _meta_for(payload: dict[str, Any]) -> dict[str, Any] | None: return {"raven.target": target} -def translate(event: Any, *, cwd: str | None = None) -> Translated: +def translate(event: Any, *, cwd: str | None = None, session_cost: float | None = None) -> Translated: """Read one wire event. Pure: no state, no I/O, no event loop. Kept pure so the frames it produces can be validated against the official @@ -207,6 +207,9 @@ def translate(event: Any, *, cwd: str | None = None) -> Translated: An event this has no branch for yields an empty result rather than raising. A translator that crashed on an unrecognised event would take the whole connection down over a wire event somebody added for the web client. + + ``session_cost`` is the session's cumulative cost, which the caller keeps: + it is what ACP's ``usage_update.cost`` reports, and no one event carries it. """ if not isinstance(event, dict): return Translated() @@ -242,7 +245,7 @@ def translate(event: Any, *, cwd: str | None = None) -> Translated: # latched one earlier; ``end_turn`` here is the default that a plain # completion means. The usage rides out first, so the client has the # turn's cost before the turn is over rather than after. - usage = _usage_update(payload.get("usage")) + usage = _usage_update(payload.get("usage"), session_cost) return Translated(updates=() if usage is None else (usage,), stop="end_turn") if kind == "media": @@ -361,8 +364,8 @@ def _file_uri(path: str) -> str: return "file://" + quote(path, safe="/") -def _usage_update(usage: Any) -> dict[str, Any] | None: - """How full the context window is, and what the turn cost. +def _usage_update(usage: Any, session_cost: float | None = None) -> dict[str, Any] | None: + """How full the context window is, and what the session has cost so far. The one place the rich accounting already on the internal wire maps cleanly onto ACP: ``used`` and ``size`` are the context numbers, and ``Cost`` wants @@ -373,8 +376,11 @@ def _usage_update(usage: Any) -> dict[str, Any] | None: no usage at all (a cached reply, a hook short-circuit) has nothing to say here -- an update of zeroes is not the same statement as no update. - The currency is USD because cost_usd carries only verified dollar amounts - reported by the provider, never a locally estimated price. + ``cost`` is ``session_cost``, not the turn's ``cost_usd``: ACP defines it as + the session's cumulative cost, and the turn's alone made a client show the + last turn's spend as the whole session's. The currency is USD because the + total sums only dollar amounts providers reported, never a locally estimated + price. """ if not isinstance(usage, dict): return None @@ -383,12 +389,27 @@ def _usage_update(usage: Any) -> dict[str, Any] | None: if not isinstance(used, int) or not isinstance(size, int) or size <= 0 or used < 0: return None update: dict[str, Any] = {"sessionUpdate": "usage_update", "used": used, "size": size} - cost = usage.get("cost_usd") - if isinstance(cost, (int, float)) and not isinstance(cost, bool) and cost >= 0: - update["cost"] = {"amount": float(cost), "currency": "USD"} + if _usd(session_cost) is not None: + update["cost"] = {"amount": float(session_cost), "currency": "USD"} return update +def _usd(value: Any) -> float | None: + """``value`` as a reported dollar amount, or None when it is not one.""" + if isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0: + return float(value) + return None + + +def _turn_spend(event: Any) -> float | None: + """What the turn a ``message.complete`` ends reported spending, or None for any other event.""" + if not isinstance(event, dict) or event.get("type") != "message.complete": + return None + payload = event.get("payload") + usage = payload.get("usage") if isinstance(payload, dict) else None + return _usd(usage.get("cost_usd")) if isinstance(usage, dict) else None + + def _clip_deep(value: Any, depth: int = 0) -> Any: """Clip every string in a JSON-shaped value, structure intact. @@ -698,6 +719,13 @@ class AcpSession: cwd: str subscription_id: str | None = None turn: _Turn | None = None + cost_usd: float | None = None + """What the session has cost so far, in provider-reported USD: what it had + spent when it was opened here, then each turn's spend added as it ends. None + until some call reports a price.""" + cost_known: bool = True + """False when what the session had spent before it was opened here could not + be read: a total counted from zero would understate it, so none is sent.""" class UpdateTranslator: @@ -919,7 +947,9 @@ async def send_frame(self, frame: Any) -> None: await self._deliver(session, params.get("event")) async def _deliver(self, session: AcpSession, event: Any) -> None: - result = translate(event, cwd=session.cwd) + if (spent := _turn_spend(event)) is not None: + session.cost_usd = (session.cost_usd or 0.0) + spent + result = translate(event, cwd=session.cwd, session_cost=session.cost_usd if session.cost_known else None) # Held thoughts go out before anything that has to follow them: a # chunk of the answer, a tool call, an ending. An event that carries # none of those (a thought, or nothing at all) leaves them held, which diff --git a/raven/acp_client/acp_agent.py b/raven/acp_client/acp_agent.py index a7884390a..b9c2eeed2 100644 --- a/raven/acp_client/acp_agent.py +++ b/raven/acp_client/acp_agent.py @@ -37,7 +37,7 @@ from loguru import logger from raven.acp_client import remote, remote_runs -from raven.acp_client.acp_dialects import AcpDialect, ToolCall, content_texts, dialect_for +from raven.acp_client.acp_dialects import AcpDialect, ToolCall, agent_version, content_texts, dialect_for from raven.acp_client.ask_user import AskUserResponder, clarify_responder from raven.acp_client.capabilities import ( CapabilitySnapshot, @@ -57,7 +57,7 @@ AcpRemoteError, AcpTimeoutError, ) -from raven.agent.subagent import activity +from raven.agent.subagent import activity, github_copilot from raven.agent.subagent.attachments import attachment_blocks from raven.agent.subagent.backends import turn_rows from raven.agent.subagent.backends.base import bounded_delta, clamp_output @@ -161,9 +161,73 @@ class AcpEmptyTurnError(RuntimeError): """ +class AcpProviderRefusalError(AcpEmptyTurnError): + """The agent ended its turn normally and its whole reply was its model provider's refusal. + + ``said`` is that reply as the agent wrote it, kept apart from raven's reading + of it so a caller that explains refusals itself (the connect) reads the + agent's words rather than this error's. + """ + + def __init__(self, message: str, *, said: str) -> None: + super().__init__(message) + self.said = said + + _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. + + ``counts`` is the turn's ``usage`` in the field names ACP's draft gives it. + Its ``inputTokens`` leaves out the cached reads and writes reported beside it + (claude-agent-acp and codex-acp both), so they are kept apart rather than + taken out again. ``None`` when the input and output counts are not both + there: the ledger stores numbers, and a row of zeros would say the turn read + and wrote nothing. + """ + if not isinstance(counts, dict): + return None + read, wrote = counts.get("inputTokens"), counts.get("outputTokens") + if not (_counted(read) and _counted(wrote)): + 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, + } + + +def _turn_cost(reported: Any, spent_before: float | None) -> tuple[float | None, float | None]: + """What one turn cost in USD, and the session's cumulative cost to measure the next from. + + ACP's ``usage_update.cost`` is the session's cumulative cost, so a turn + costs what it added to it. ``spent_before`` is the total at the turn's + start: 0 for a session this process opened, ``None`` when it is unknown -- + a session resumed after this process started -- and then the turn's cost is + unknown too, not the whole total charged again. A total that went down says + 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: + return None, spent_before + if spent_before is None or amount < spent_before: + return None, float(amount) + return float(amount) - spent_before, float(amount) + + def _holds_session(exc: AcpRemoteError) -> bool: """Whether the agent refused ``session/load`` because this process already holds the session. @@ -272,6 +336,7 @@ def __init__( workspace: Path | None = None, limit: int | None = None, track_files: bool = True, + agent_version: str | None = None, ) -> None: # The reply's cap, renewed for every message rather than spent across # the turn: the reply is one message, and narration spending a shared @@ -287,6 +352,7 @@ def __init__( # The spec by default, so a collector built without one (every test that # only cares about answer text) still reads tool calls correctly. self._dialect = dialect or AcpDialect() + self._agent_version = agent_version self._tool_ran = False self._steered = False # Where each message starts in `answer`. Only a step opens one: the @@ -792,6 +858,19 @@ def _plan(self, update: dict[str, Any]) -> None: return self.events.append({"t": "result", "id": _PLAN_CALL_ID, "ok": True, "text": checklist, "at": self._now()}) + def turn_counts(self, answered: Any) -> dict[str, Any]: + """The turn's token counts, from one source: never both, as ``note_usage`` adds. + + ``answered`` is the prompt response's ``usage``, read where the dialect + says it counts the whole turn; the ``usage_update`` then carries only the + window and the cost. Otherwise the update, where an earlier agent was seen + to send its counts. + """ + counts_the_turn = self._dialect.prompt_usage_counts_the_turn(self._agent_version) + if counts_the_turn and isinstance(answered, dict) and answered: + return answered + return self.usage + @property def calls(self) -> list[ToolCall]: return [ev["call"] for ev in self.events if ev["t"] == "call"] @@ -1051,6 +1130,7 @@ def __init__( pool: Any = None, machine: str | None = None, remote_cwd: str | None = None, + preset: str | None = None, ) -> None: self.name = name self.command = command @@ -1059,6 +1139,8 @@ def __init__( # see ``raven.acp_client.remote`` for what changes when it is set. self.machine = (machine or "").strip() or None self.remote_cwd = remote_cwd + # Which shipped preset this row is, for the readings measured per agent. + self.preset = preset self.env = env or {} self.ready_timeout_ms = ready_timeout_ms self.timeout = timeout @@ -1075,6 +1157,17 @@ def __init__( # route into a session -- and on every fresh session a spawn opens -- # is said once per value rather than once per turn. self._model_refused: set[str] = set() + # An instance whose earlier session the agent could not resume, by + # (session key, handle): the next reply on that handle has to say it + # starts over. Held until a turn there answers, not only for the turn + # that met the refusal -- the binding is dropped at that refusal, so a + # retry after a failed turn opens a fresh session with nothing of its + # own to report. For this process's life, as the backend is. + self._resume_lost: dict[tuple[str, str], str] = {} + # Each session's cumulative cost as last reported, for the cost of its + # next turn (`_turn_cost`). Only for sessions this process has seen, and + # only on an agent that resumes them: no other session has a next turn. + self._session_spend: dict[str, float] = {} self._registry = registry or get_registry() # Which pool this backend's turns are served from. Almost always the # process-wide one, and held unresolved until it is used so that @@ -1766,6 +1859,7 @@ async def run( workspace=Path(session_cwd) if launched is None else None, limit=self.max_output_chars, track_files=launched is None, + agent_version=agent_version(connection.initialize), ) # Built here, in the turn's context, for the reason `_TurnCollector` # documents: the read loop's ContextVars predate this run, and the @@ -1898,7 +1992,25 @@ async def run( activity.note_output_limit() text = collector.text frames = self._frames(journal, connection, session_id, frames_start) - self._record(span, collector, stop_reason=stop_reason, started=started, frames=frames) + self._record( + span, + collector, + stop_reason=stop_reason, + started=started, + frames=frames, + counted=result.get("usage") if isinstance(result, dict) else None, + ) + # Before the checks below: a turn that answered nothing usable + # still spent what it spent. + await self._book_usage( + collector, + result.get("usage") if isinstance(result, dict) else None, + initialize=connection.initialize, + session_id=session_id, + resumed=resumed, + session_key=session_key, + task_id=task_id, + ) if not text: tail = client.stderr_tail(1500) @@ -1931,6 +2043,10 @@ async def run( f"{detail.strip() or ''}" ) + if (refused := self._provider_refusal(text)) is not None: + span.error("the reply is the model provider's refusal") + raise AcpProviderRefusalError(refused, said=text) + if not resumed and self.is_stateful: # Deferred commit, as the cli transport does it: binding a handle # before the first turn succeeded would resume a session that @@ -1940,7 +2056,7 @@ async def run( await self._registry.commit(skey, self.name, handle, session_id, kind="acp") reply = await self._finished(collector.reply, stop_reason=stop_reason, span=span, sink=on_delta) - notes = [self._mcp_note(grant, lent)] + notes = [self._resume_lost.pop((skey, handle), None), self._mcp_note(grant, lent)] if launched is not None: # Where to find what it made: the caller's workspace line names # this computer, and what it wrote is there, by address -- @@ -2083,6 +2199,54 @@ async def steer(text: str) -> str: return steer + async def _book_usage( + self, + collector: _TurnCollector, + answered: Any, + *, + initialize: Any, + session_id: str, + resumed: bool, + session_key: str | None, + task_id: str, + ) -> None: + """Book the turn's tokens and cost in the usage ledger, as an in-process sub-agent's calls are. + + Not for a raven agent: it books its own calls under the delegating + conversation (the prompt's ``raven.usage``), so booking its turn here as + well would count it twice. + """ + info = initialize.get("agentInfo") if isinstance(initialize, dict) else None + if isinstance(info, dict) and info.get("name") == "raven": + return + before = self._session_spend.get(session_id, None if resumed else 0.0) + cost, spent = _turn_cost(collector.usage, before) + if spent is not None and self.is_stateful: + self._session_spend[session_id] = spent + counts = _ledger_counts(collector.turn_counts(answered)) + if counts is None and cost is None: + return + model = self._model_pushed.get(session_id) or self._model_baseline.get(session_id) or "" + await activity.note_ledger_usage( + {**(counts or {}), "cost_usd": cost}, + model=model, + session_key=session_key, + task_id=task_id, + counts_known=counts is not None, + ) + + def _provider_refusal(self, reply: str) -> str | None: + """Why the run failed, when the whole reply is the agent's model provider refusing the call. + + Copilot ends a refused model call as a finished turn whose message is the + refusal, so neither an empty-turn check nor a failed-call check sees it, + and handed back it is a completed run whose output is the error. Read only + for the agent it was measured on. + """ + if self.preset != "github_copilot" or (refused := github_copilot.read_reply(reply)) is None: + return None + return f"acp agent {self.name!r} answered with its model provider's refusal: {refused[0]}" + async def _finished( self, text: str, @@ -2177,6 +2341,7 @@ async def request_or_busy(method: str, params: dict[str, Any]) -> Any: f"the same silence again." ) from None + lost: str | None = None if self.is_stateful: known = await self._registry.lookup(skey, self.name, handle, kind="acp") if known is not None and (self._snapshot and self._snapshot.can_load): @@ -2221,6 +2386,7 @@ async def request_or_busy(method: str, params: dict[str, Any]) -> Any: exc.message, ) await self._registry.unbind(skey, self.name, handle) + lost = exc.message except AcpError: # A transport-level failure is not evidence the session was # pruned, so the binding is kept and the caller is told. @@ -2234,6 +2400,11 @@ async def request_or_busy(method: str, params: dict[str, Any]) -> Any: self._note_session_model(session_id, result) await self._set_mode(client, session_id, mode, budget=budget) await self._set_model(client, session_id, session_model, budget=budget) + if lost is not None: + self._resume_lost[(skey, handle)] = ( + f"Instance {handle!r} could not pick up its earlier conversation ({lost}), " + "so this reply comes from a fresh session that has none of it" + ) return session_id, False async def _set_model(self, client: Any, session_id: str, model: str | None, *, budget: float) -> None: @@ -2393,6 +2564,7 @@ def _record( stop_reason: Any, started: float, frames: dict[str, Any], + counted: Any = None, ) -> None: counts = collector.counts() record_events(span, transport="acp", kinds=counts) @@ -2413,7 +2585,7 @@ def _record( # already published as they landed; this is the settled list, with the # labels every later frame corrected, written over it rather than added. activity.set_tool_calls(activity.current(), collector.tool_calls, collector.failed_calls) - activity.note_usage(collector.usage) + activity.note_usage(collector.turn_counts(counted)) activity.note_steps(counts) activity.note_thoughts(thought_chars) activity.note_transcript(collector.messages()) diff --git a/raven/acp_client/acp_dialects/__init__.py b/raven/acp_client/acp_dialects/__init__.py index 19c9848c7..261a910b9 100644 --- a/raven/acp_client/acp_dialects/__init__.py +++ b/raven/acp_client/acp_dialects/__init__.py @@ -26,12 +26,21 @@ def dialect_for(initialize: Any) -> AcpDialect: """The dialect for the agent that answered this ``initialize``.""" - payload = initialize if isinstance(initialize, dict) else {} - info = payload.get("agentInfo") - name = (info or {}).get("name") if isinstance(info, dict) else None + name = _agent_info(initialize).get("name") return _for_name(name if isinstance(name, str) else "") +def agent_version(initialize: Any) -> str | None: + """The version the agent that answered this ``initialize`` reports, if it reports one.""" + version = _agent_info(initialize).get("version") + return version if isinstance(version, str) else None + + +def _agent_info(initialize: Any) -> dict[str, Any]: + info = initialize.get("agentInfo") if isinstance(initialize, dict) else None + return info if isinstance(info, dict) else {} + + def _for_name(agent_name: str) -> AcpDialect: """The dialect whose key appears in ``agent_name``, else the spec.""" lowered = agent_name.lower() @@ -47,6 +56,7 @@ def _for_name(agent_name: str) -> AcpDialect: "CodexDialect", "ToolCall", "DialectResult", + "agent_version", "content_texts", "dialect_for", ] diff --git a/raven/acp_client/acp_dialects/base.py b/raven/acp_client/acp_dialects/base.py index c37b7d58b..22cdd1072 100644 --- a/raven/acp_client/acp_dialects/base.py +++ b/raven/acp_client/acp_dialects/base.py @@ -185,6 +185,16 @@ class AcpDialect: adapter that does not overrides this, and the run's file record then has to tell a creation from a rewrite some other way.""" + def prompt_usage_counts_the_turn(self, version: str | None) -> bool: + """Whether the prompt response's ``usage`` counts the whole turn, for the adapter at ``version``. + + The turn's token counts are read from it when it does. claude-agent-acp + sums every call of the turn into it; an adapter that reports something + narrower overrides this, and its counts are then left unread rather than + recorded as the turn's. ``version`` is the one ``agentInfo`` reports. + """ + return True + def tool_name(self, update: dict[str, Any]) -> str: """The transport's own name for the call, at the finest grain it gives. diff --git a/raven/acp_client/acp_dialects/codex.py b/raven/acp_client/acp_dialects/codex.py index 77b7b0c30..1cba26616 100644 --- a/raven/acp_client/acp_dialects/codex.py +++ b/raven/acp_client/acp_dialects/codex.py @@ -99,17 +99,36 @@ def _file_change_paths(content: Any) -> str: # the transcript as garbage. Tab and newline are the two that carry meaning. _CONTROL = re.compile(r"[\x00-\x08\x0b-\x0c\x0e-\x1f]") +_RELEASE = re.compile(r"(\d+)\.(\d+)\.(\d+)") +_WHOLE_TURN_USAGE = (2, 2, 1) + def _clean(text: str) -> str: return _CONTROL.sub("", text.replace("\r\n", "\n").replace("\r", "\n")) +def _release(version: str | None) -> tuple[int, int, int] | None: + match = _RELEASE.match(version) if isinstance(version, str) else None + if match is None: + return None + major, minor, patch = match.groups() + return int(major), int(minor), int(patch) + + class CodexDialect(AcpDialect): key = "codex-acp" plan_tool_name = "update_plan" """codex's own name for the tool behind a plan frame (``plan_tool.rs``).""" + def prompt_usage_counts_the_turn(self, version: str | None) -> bool: + """From 2.2.1 the prompt response's ``usage`` is the thread's total at the + turn's end less its total at the start. 2.1.1 and earlier built it from + the app server's ``tokenUsage.last``: the turn's final model call, not the + turn. A version that is no release number is read as the older kind.""" + release = _release(version) + return release is not None and release >= _WHOLE_TURN_USAGE + def tool_name(self, update: dict[str, Any]) -> str: raw = _dict(update.get("rawInput")) meta = _dict(update.get("_meta")) diff --git a/raven/acp_client/remote_resume.py b/raven/acp_client/remote_resume.py index eba2bc2e5..ab4d6cbc6 100644 --- a/raven/acp_client/remote_resume.py +++ b/raven/acp_client/remote_resume.py @@ -123,6 +123,8 @@ 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: + return PickedUp("error", refused, stop_reason) reply = await backend._finished(collector.reply, stop_reason=stop_reason, span=_Span(), sink=None) reply += ( f"\n\n[raven] Ran on machine {target.label}, in {remote.shown(target, run.session_cwd)}, " diff --git a/raven/agent/loop/turn_path.py b/raven/agent/loop/turn_path.py index e6c090732..4e02d71ae 100644 --- a/raven/agent/loop/turn_path.py +++ b/raven/agent/loop/turn_path.py @@ -2818,6 +2818,12 @@ def _rerun_pending(final_content: str | None, msgs: list[dict], status: str) -> # 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 + # The same slot for a turn its own budget ended -- the tool cap, the + # wall clock or a stalled tool, which is all ``interrupted`` means. + # Its wrap-up reads as an ordinary answer, so a surface that reports + # the turn (ACP's stop reason) has only this to tell it from one + # that finished. Written or cleared on the same terms as above. + session.metadata["budget_stop_turn_at"] = _turn_start() if outcome.status == "interrupted" 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/agent/subagent/activity.py b/raven/agent/subagent/activity.py index 2bf0c13d0..7156a720c 100644 --- a/raven/agent/subagent/activity.py +++ b/raven/agent/subagent/activity.py @@ -637,12 +637,28 @@ async def note_provider_usage( ) -> None: """Record one direct provider call in the run activity and usage ledger.""" note_usage(usage) + await note_ledger_usage(usage, model=model, session_key=session_key, task_id=task_id) + + +async def note_ledger_usage( + usage: Any, + *, + model: str, + session_key: str | None, + task_id: str, + counts_known: bool = True, +) -> None: + """Book one delegated call in the usage ledger under the conversation that delegated it. + + ``counts_known`` false books a call whose token counts are unknown: they are + stored as missing and only its cost is booked. + """ if session_key is None or not isinstance(usage, dict): return try: from raven.token_wise.usage_tracker import record_delegated_usage - await record_delegated_usage(usage, model=model, root=session_key, task_id=task_id) + await record_delegated_usage(usage, model=model, root=session_key, task_id=task_id, counts_known=counts_known) except Exception: # noqa: BLE001 - accounting must never fail the delegated run logger.debug("subagent activity: usage ledger write failed", exc_info=True) diff --git a/raven/agent/subagent/backends/__init__.py b/raven/agent/subagent/backends/__init__.py index 58b76f1a3..e78620804 100644 --- a/raven/agent/subagent/backends/__init__.py +++ b/raven/agent/subagent/backends/__init__.py @@ -442,6 +442,7 @@ def build_third_party_backend( pool=pool, machine=getattr(cfg, "machine", None), remote_cwd=getattr(cfg, "remote_cwd", None), + preset=getattr(cfg, "preset", None), ) if kind == "openai": return OpenAIApiBackend( diff --git a/raven/agent/subagent/backends/raven_loop.py b/raven/agent/subagent/backends/raven_loop.py index e7a96d53f..93d94388d 100644 --- a/raven/agent/subagent/backends/raven_loop.py +++ b/raven/agent/subagent/backends/raven_loop.py @@ -549,6 +549,9 @@ def participant_step(phase: str, *, response: Any = None) -> StepView: # running out of rounds: the one wrap-up call below is worded for each, # and so is the failure if that call answers nothing either. answered_blank = False + # Whether the rounds ran out with the model still calling tools: the + # answer is then a wrap-up, and the reply says it is partial. + ran_out = False # Whether the LAST model response of this run was cut at the output # ceiling, not whether any was: a round that was cut and then answered # in full delivered its answer, and the verdict reading this is asking @@ -827,6 +830,7 @@ def participant_step(phase: str, *, response: Any = None) -> StepView: # another, it answers from what it has. Observed need -- a research # node spent all fifteen rounds on web_fetch and wrote nothing, and # the run's whole output was the placeholder that used to sit here. + ran_out = not answered_blank if answered_blank: logger.warning("Subagent [{}] ended its turn with an empty reply; asking once with no tools", task_id) reason = "Your reply was empty." @@ -896,12 +900,28 @@ def participant_step(phase: str, *, response: Any = None) -> StepView: # anything has streamed the turn sends no closing text, so an answer # kept from ``on_delta`` would be recorded and never shown. await on_delta(final_result) + # What the instance's own history keeps as its reply. + kept = final_result + if ran_out: + # The ACP backend's partial-turn notice for the same case, so a caller + # reads a capped run the same way whichever backend ran it. Not kept: + # it is raven telling the caller about this run, and a resumed + # instance that read it as its own words could repeat it in a reply + # that was not cut. + notice = ( + f"\n\n[raven] The sub-agent used all {self._MAX_ITERATIONS} tool rounds before finishing; " + "reply is partial." + ) + if on_delta is not None: + await on_delta(notice) + final_result += notice if note := grant.note_text(): notice = f"\n\n[raven] {note}." if on_delta is not None: await on_delta(notice) final_result += notice + kept += notice if on_messages is not None: - messages.append({"role": "assistant", "content": final_result}) + messages.append({"role": "assistant", "content": kept}) on_messages(messages) return final_result diff --git a/raven/agent/subagent/github_copilot.py b/raven/agent/subagent/github_copilot.py index 61eae41f7..40dd04404 100644 --- a/raven/agent/subagent/github_copilot.py +++ b/raven/agent/subagent/github_copilot.py @@ -84,3 +84,40 @@ def read(said: str) -> tuple[str, Remedy | None] | None: if _SERVER.search(said): return (f"its model provider returned a server error. It said: {said}", None) return None + + +# How each sentence above opens, as Copilot writes it. A task's answer can open +# with an error it found ("Error: 502 Bad Gateway from staging"); Copilot's own +# refusals are these fixed sentences, and nothing else is read as one. +_REFUSAL_OPENING = re.compile( + r"Error: (?:" + r"Authentication failed with provider at " + r"|\d{3} Your credit balance is too low" + r"|\d{3} Your Copilot subscription has expired" + r"|Model '[^']+' not found on provider at " + r"|Could not connect to local model provider" + r"|Execution failed: Failed to get response from the AI model" + r"|Failed to get response from the AI model; retried" + r")", + re.IGNORECASE, +) +# Longer than any refusal above: the longest measured runs to under 300 characters. +_REFUSAL_MAX_CHARS = 1000 + + +def read_reply(reply: str) -> tuple[str, Remedy | None] | None: + """:func:`read`, for the answer to a dispatched task rather than to a ping. + + Every refusal above came back as the whole assistant message. A real answer + can cite the same words -- a 500 in the log of a deploy it is debugging, an + error a request it was asked to make returned -- and failing it for that + would lose the work, so a reply is read only when it opens the way one of + Copilot's own refusals does, and only when it is as short as one: a long + reply is work, and the patterns above are not run over text that size. A + ping's reply answers a one-word prompt, which is why the connect reads it as + it is. + """ + said = reply.strip() + if len(said) > _REFUSAL_MAX_CHARS or not _REFUSAL_OPENING.match(said): + return None + return read(said) diff --git a/raven/agent/subagent/probe.py b/raven/agent/subagent/probe.py index 1c2da7460..3f58647a8 100644 --- a/raven/agent/subagent/probe.py +++ b/raven/agent/subagent/probe.py @@ -223,8 +223,12 @@ def _said(exc: BaseException) -> tuple[str, str | None]: when no agent answered at all (a timeout, a process that never started), and the whole text is then all there is to go on. """ + from raven.acp_client.acp_agent import AcpProviderRefusalError from raven.acp_client.protocol import reason_of, remote_error_in + if isinstance(exc, AcpProviderRefusalError): + # The run already read the refusal; its words are the agent's answer. + return str(exc), exc.said shown = str(exc) error = remote_error_in(exc) if error is None: diff --git a/raven/session/manager.py b/raven/session/manager.py index 3dcb44f24..9130113ef 100644 --- a/raven/session/manager.py +++ b/raven/session/manager.py @@ -345,12 +345,29 @@ def clear(self) -> None: # rule in ``SessionManager._metadata_to_write``: a key this copy merely # omits comes back on the next save. self.metadata["annotated_through"] = 0 + self._forget_turn_markers(0) self.truncations += 1 self.log_id = uuid.uuid4().hex self.log_generation = 0 self.log_started_at = datetime.now(timezone.utc).isoformat(timespec="microseconds") self.updated_at = datetime.now() + 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 + 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"): + at = self.metadata.get(key) + if isinstance(at, int) and at >= from_index: + self.metadata[key] = None + def undo_last_turn(self, n: int = 1) -> int: """Drop the last ``n`` user-turn blocks from the unconsolidated tail. @@ -373,6 +390,7 @@ def undo_last_turn(self, n: int = 1) -> int: cursor = self.metadata.get("annotated_through") if isinstance(cursor, int) and cursor > cut_index: self.metadata["annotated_through"] = cut_index + self._forget_turn_markers(cut_index) self.truncations += 1 self.log_generation += 1 self.updated_at = datetime.now() diff --git a/raven/token_wise/usage_tracker.py b/raven/token_wise/usage_tracker.py index 85878b090..981ac65bd 100644 --- a/raven/token_wise/usage_tracker.py +++ b/raven/token_wise/usage_tracker.py @@ -126,14 +126,22 @@ async def record_delegated_usage( model: str, root: str, task_id: str, + counts_known: bool = True, ) -> None: - """Persist one in-process sub-agent call under its delegation root.""" + """Persist one delegated call under its delegation root. + + ``counts_known`` false stores the call's input and output counts as missing + rather than the zeros ``normalize_usage`` reads an absent count as: a zero + says the call read and wrote nothing. + """ if not root: return from raven.providers.usage import normalize_usage normalized = normalize_usage(usage) normalized.pop("total_tokens") + if not counts_known: + normalized.update(input_tokens=None, output_tokens=None) tracker = UsageTracker() await tracker.after_llm_call( {}, diff --git a/tests/acp_stub_server.py b/tests/acp_stub_server.py index ca97a9b96..92c66d634 100644 --- a/tests/acp_stub_server.py +++ b/tests/acp_stub_server.py @@ -8,6 +8,13 @@ Behaviour is chosen by ``ACP_STUB_MODE``: - ``ok`` - full handshake, one text chunk plus a tool call, then end_turn. + ``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_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. - ``reject_init`` - answers ``initialize`` with a JSON-RPC error. - ``no_session`` - handshake fine, ``session/new`` errors (an auth-shaped message). - ``no_session_sdk`` - like ``no_session``, but refused the way both ACP SDKs wrap @@ -174,9 +181,14 @@ if MODE != "no_session_delete": _SESSION_CAPS["delete"] = {} +_AGENT_VERSION = os.environ.get("ACP_STUB_AGENT_VERSION", "9.9.9") + CAPABILITIES = { "protocolVersion": 1, - "agentInfo": {"name": "stub-agent", "version": "9.9.9"}, + "agentInfo": { + "name": os.environ.get("ACP_STUB_AGENT_NAME", "stub-agent"), + **({"version": _AGENT_VERSION} if _AGENT_VERSION else {}), + }, "agentCapabilities": { "loadSession": True, "promptCapabilities": {"image": True}, @@ -330,6 +342,10 @@ 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"}) + return if MODE == "dies_mid_turn": # Half an answer, then the process is gone: the prompt went out and # fails on the client side with the connection, not with a reply. @@ -696,9 +712,17 @@ def handle_prompt(request_id, params) -> None: "content": [{"type": "content", "content": {"type": "text", "text": "the file says hello"}}], }, ) - update(session_id, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "pong"}}) - update(session_id, {"sessionUpdate": "usage_update", "size": 1000, "used": 42}) - ok(request_id, {"stopReason": "end_turn"}) + reply = os.environ.get("ACP_STUB_REPLY", "pong") + update(session_id, {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": reply}}) + reported = json.loads(os.environ.get("ACP_STUB_USAGE_UPDATE", '{"size": 1000, "used": 42}')) + if per_prompt := os.environ.get("ACP_STUB_COST_PER_PROMPT"): + # 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) def handle_cancel(params) -> None: diff --git a/tests/test_acp_dialects.py b/tests/test_acp_dialects.py index 37a4dae0a..8e8e88f67 100644 --- a/tests/test_acp_dialects.py +++ b/tests/test_acp_dialects.py @@ -12,6 +12,8 @@ import json +import pytest + from raven.acp_client.acp_dialects import AcpDialect, ClaudeCodeDialect, CodexDialect, dialect_for @@ -24,6 +26,32 @@ def test_the_dialect_is_chosen_from_the_agents_own_handshake() -> None: assert isinstance(codex, CodexDialect) +@pytest.mark.parametrize( + ("version", "counts_the_turn"), + [ + ("2.2.1", True), + ("2.2.1-preview.1", True), + ("2.3.0", True), + ("10.0.0", True), + ("2.1.2-preview.7", False), + ("2.1.1", False), + ("1.13.1", False), + ("", False), + (None, False), + ], +) +def test_codex_prompt_usage_counts_the_turn_from_2_2_1(version: str | None, counts_the_turn: bool) -> None: + """2.2.1 builds the prompt response's ``usage`` from the thread's total at the + turn's end less its total at the start; 2.1.1 and earlier from ``tokenUsage.last``, + the turn's final model call. A version that is no release number is the older kind. + + A pre-release reads as its release. 2.2.1's previews already counted the turn; + the last previews of 2.1.2, never released, did too and read as the older + kind, which leaves their counts unknown rather than wrong.""" + assert CodexDialect().prompt_usage_counts_the_turn(version) is counts_the_turn + assert AcpDialect().prompt_usage_counts_the_turn(version) is True + + def test_an_unmeasured_adapter_gets_the_spec_and_not_a_guess() -> None: """OpenCode reported no tool call in any capture, so it has no dialect. diff --git a/tests/test_acp_methods.py b/tests/test_acp_methods.py index d0831694a..14f1595d9 100644 --- a/tests/test_acp_methods.py +++ b/tests/test_acp_methods.py @@ -64,6 +64,15 @@ def __init__(self) -> None: self.dispatcher.register("session.resume", self._resume) self.dispatcher.register("model.options", self._model_options) self.dispatcher.register("config.set", self._config_set) + # What ``session.usage`` reports a session to have spent; an exception is raised. + self.usage_result: dict | Exception = {"cost_usd": None} + self.dispatcher.register("session.usage", self._usage) + + async def _usage(self, params: dict) -> dict: + self.calls.append(("session.usage", params)) + if isinstance(self.usage_result, Exception): + raise self.usage_result + return self.usage_result async def _model_options(self, params: dict) -> dict: self.calls.append(("model.options", params)) @@ -574,6 +583,42 @@ async def test_the_same_refusals_as_a_fresh_session_apply(self, rig): assert "error" not in servers, servers +class TestSessionCost: + """``usage_update.cost`` is the session's cumulative cost, so a session + opened here starts from what it had already spent: the same figure the + session's own usage panel reports, which outlives the process.""" + + async def _opened(self, rig, method: str): + rig.stack.stored["acp:old"] = [{"role": "user", "text": "hi"}] + await rig.handshake() + await rig.call(method, {"sessionId": "acp:old", "cwd": str(rig.tmp_path / "project"), "mcpServers": []}) + return rig.translator.get("acp:old") + + @pytest.mark.parametrize("method", ["session/load", "session/resume"]) + async def test_a_session_opened_here_starts_from_what_it_had_spent(self, rig, method): + rig.stack.usage_result = {"cost_usd": 1.5} + + session = await self._opened(rig, method) + + assert (session.cost_usd, session.cost_known) == (1.5, True) + assert ("session.usage", {"session_id": "acp:old"}) in rig.stack.calls + + async def test_a_session_whose_spend_cannot_be_read_has_no_total(self, rig): + rig.stack.usage_result = RuntimeError("telemetry unreadable") + + session = await self._opened(rig, "session/load") + + assert session.cost_known is False + + async def test_a_new_session_has_spent_nothing_and_asks_nobody(self, rig): + await rig.handshake() + response = await rig.call("session/new", {"cwd": str(rig.tmp_path / "project"), "mcpServers": []}) + + session = rig.translator.get(response["result"]["sessionId"]) + assert (session.cost_usd, session.cost_known) == (None, True) + assert not any(name == "session.usage" for name, _ in rig.stack.calls) + + class TestSessionResume: """``session/resume`` continues a session a client already knows about. @@ -2237,6 +2282,33 @@ def test_a_more_specific_stop_reason_is_never_overwritten_by_the_ceiling(): assert AcpMethods._stop_reason(stop, sessions, "k", 0) == stop +def test_a_turn_cut_by_its_own_budget_is_reported_as_max_turn_requests(): + """The loop's tool, clock or stalled-tool budget ended the turn and a wrap-up + answered for it. ``end_turn`` would claim the turn finished; the schema's own + word for a turn its budget ended is ``max_turn_requests``.""" + sessions = _MetaSessions([{"role": "user", "content": "hi"}], {"budget_stop_turn_at": 0}) + assert AcpMethods._stop_reason("end_turn", sessions, "k", 0) == "max_turn_requests" + + +def test_a_budget_stop_left_by_an_earlier_turn_does_not_answer_this_prompt(): + messages = [{"role": "user", "content": str(n)} for n in range(6)] + sessions = _MetaSessions(messages, {"budget_stop_turn_at": 0}) + assert AcpMethods._stop_reason("end_turn", sessions, "k", 5) == "end_turn" + + +def test_a_more_specific_stop_reason_is_never_overwritten_by_the_budget(): + sessions = _MetaSessions([{"role": "user", "content": "hi"}], {"budget_stop_turn_at": 0}) + for stop in ("cancelled", "refusal"): + assert AcpMethods._stop_reason(stop, sessions, "k", 0) == stop + + +def test_a_budget_stop_whose_wrap_up_was_also_cut_reports_the_budget(): + """The budget is what ended the turn; the wrap-up being cut as well does not + make the turn any less over budget.""" + sessions = _MetaSessions([{"role": "user", "content": "hi"}], {"budget_stop_turn_at": 0, "output_limit_turn_at": 0}) + assert AcpMethods._stop_reason("end_turn", sessions, "k", 0) == "max_turn_requests" + + def test_turn_meta_reads_only_what_hooks_filed(): """Back to the documented contract: the table is the hooks' stash and nothing else, so nothing the host decides from rides in it.""" diff --git a/tests/test_acp_updates.py b/tests/test_acp_updates.py index 72d414cd7..4f9635251 100644 --- a/tests/test_acp_updates.py +++ b/tests/test_acp_updates.py @@ -17,6 +17,7 @@ import asyncio import json +from typing import Any import pytest @@ -388,7 +389,7 @@ def test_a_malformed_target_is_ignored_rather_than_forwarded(self): class TestUsage: """The one place raven's rich accounting maps cleanly onto ACP.""" - def test_a_completion_carries_the_window_and_the_cost(self): + def test_a_completion_carries_the_window_and_the_sessions_cost(self): result = translate( { "type": "message.complete", @@ -397,12 +398,13 @@ def test_a_completion_carries_the_window_and_the_cost(self): "usage": { "prompt_tokens": 100, "completion_tokens": 20, - "cost_usd": 0.0042, + "cost_usd": 0.001, "context_used": 13500, "context_max": 200000, }, }, - } + }, + session_cost=0.0042, ) assert result.stop == "end_turn" @@ -452,11 +454,54 @@ def test_a_cost_of_zero_is_still_reported(self): """Zero is a real answer -- a cached reply cost nothing -- and different from not knowing.""" result = translate( - {"type": "message.complete", "payload": {"usage": {"context_used": 1, "context_max": 2, "cost_usd": 0}}} + {"type": "message.complete", "payload": {"usage": {"context_used": 1, "context_max": 2, "cost_usd": 0}}}, + session_cost=0, ) assert result.updates[0]["cost"] == {"amount": 0.0, "currency": "USD"} + def test_a_turns_own_cost_is_not_the_sessions(self): + """ACP defines ``cost`` as the session's cumulative cost. The turn's cost + sent as it was made a client that follows the schema show the last + turn's spend as the whole session's.""" + result = translate( + {"type": "message.complete", "payload": {"usage": {"context_used": 1, "context_max": 2, "cost_usd": 0.5}}} + ) + + assert "cost" not in result.updates[0] + + @staticmethod + def _costs(session: AcpSession, *turn_costs: float | None) -> list[Any]: + written: list[dict] = [] + translator = UpdateTranslator(emit=written.append) + translator.add(session) + for cost in turn_costs: + usage = {"context_used": 1, "context_max": 100, **({} if cost is None else {"cost_usd": cost})} + asyncio.run(translator.send_frame(_event("message.complete", usage=usage))) + return [u.get("cost", {}).get("amount") for u in _updates(written)] + + def test_each_turn_reports_what_the_session_has_cost_so_far(self): + assert self._costs(_session(), 0.01, 0.02, 0.5) == pytest.approx([0.01, 0.03, 0.53]) + + def test_an_unpriced_turn_leaves_the_total_where_it_was(self): + """A model with no published price adds an unknown amount; the total of + 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_session_that_was_resumed_counts_on_from_its_earlier_spend(self): + session = _session() + session.cost_usd = 1.5 + + assert self._costs(session, 0.25) == pytest.approx([1.75]) + + def test_a_session_whose_earlier_spend_is_unknown_reports_no_cost(self): + """A total started at zero would understate a resumed session as surely + as the turn's own cost did; without the earlier spend there is no total.""" + session = _session() + session.cost_known = False + + assert self._costs(session, 0.25, 0.5) == [None, None] + class TestTerminationMapping: def test_a_completion_ends_the_turn(self): diff --git a/tests/test_agent_loop_max_iter_synthesis.py b/tests/test_agent_loop_max_iter_synthesis.py index 4e7b7d4aa..4df378c3c 100644 --- a/tests/test_agent_loop_max_iter_synthesis.py +++ b/tests/test_agent_loop_max_iter_synthesis.py @@ -306,3 +306,69 @@ async def test_synthesis_falls_back_to_static_on_exception(): synth = _bind_synth(provider, max_iterations=40) result = await synth([], "m", None) assert result == _MAX_ITER_STATIC_FALLBACK.format(n=40) + + +# --------------------------------------------------------------------------- # +# A turn cut by its own budget leaves a turn-scoped marker on the session # +# --------------------------------------------------------------------------- # + + +class _AnswersProvider(LLMProvider): + """Answers at once: a turn that finishes inside its budget.""" + + def __init__(self) -> None: + super().__init__(api_key="test") + + async def chat( + self, + messages, + tools=None, + model=None, + max_tokens=4096, + temperature=0.7, + reasoning_effort=None, + tool_choice=None, + ): + return LLMResponse(content="done", finish_reason="stop") + + def get_default_model(self) -> str: + return "stub" + + +def _user_turn(text: str) -> TurnRequest: + return TurnRequest( + origin=Origin.USER, + source=Source(channel="test", chat_id="c1", sender_id="user", chat_type=ChatType.DM), + text=text, + ) + + +@pytest.mark.asyncio +async def test_a_turn_that_spent_its_tool_budget_is_marked_on_the_session(workspace): + """The wrap-up reads as an ordinary answer, so the session says the turn was cut. + + Session-level for the reason ``output_limit_turn_at`` is: a surface that + reports the turn (ACP's stop reason) reads it after the turn, and stamped + with the index the turn's rows start at so an earlier turn's cut never + answers a later prompt. + """ + agent = _make_agent(workspace, _ToolThenSynthProvider()) + + out = await agent._process_message(_user_turn("do the thing"), session_key="s1") + + assert out is not None and out[0] == "Partial summary: did A and B; C is still pending." + assert agent.sessions.get_or_create("s1").metadata.get("budget_stop_turn_at") == 0 + + +@pytest.mark.asyncio +async def test_a_turn_that_finished_clears_a_budget_stop_left_by_an_earlier_one(workspace): + """Written OR cleared every turn: `/new` and undo move the message index back + without touching metadata, so an old marker could otherwise answer for a turn + that finished.""" + agent = _make_agent(workspace, _AnswersProvider()) + agent.sessions.get_or_create("s1").metadata["budget_stop_turn_at"] = 0 + + out = await agent._process_message(_user_turn("hi"), session_key="s1") + + assert out is not None and out[0] == "done" + assert agent.sessions.get_or_create("s1").metadata.get("budget_stop_turn_at") is None diff --git a/tests/test_remote_resume.py b/tests/test_remote_resume.py index 502a95445..90c9fe7aa 100644 --- a/tests/test_remote_resume.py +++ b/tests/test_remote_resume.py @@ -135,6 +135,22 @@ async def test_a_turn_cut_short_is_marked_partial_as_a_turn_ended_here_would_be( assert picked.status == "ok" and picked.reply.startswith("half\n\n[raven] stub@box stopped before finishing") +async def test_a_picked_up_copilot_reply_that_is_its_providers_refusal_is_an_error(_pool): + """The pick-up reports a left run the way the turn would have ended here, and + a Copilot turn whose whole reply is its provider's refusal fails here.""" + 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("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_session_manager.py b/tests/test_session_manager.py index 33cfc58b3..2da4d761e 100644 --- a/tests/test_session_manager.py +++ b/tests/test_session_manager.py @@ -1921,3 +1921,44 @@ def test_a_saved_time_with_an_offset_is_read_as_local_time(tmp_path: Path, offse assert loaded is not None assert loaded.updated_at.tzinfo is None assert loaded.updated_at == saved.astimezone().replace(tzinfo=None) + + +# ---- turn markers that index into the messages ------------------------------ + +_TURN_MARKERS = ("output_limit_turn_at", "budget_stop_turn_at") + + +def _session_with_two_turns_the_last_one_cut(): + from raven.session.manager import Session + + session = Session(key="tui:markers") + for role, text in (("user", "a"), ("assistant", "b"), ("user", "c"), ("assistant", "partial")): + session.add_message(role, text) + for key in _TURN_MARKERS: + session.metadata[key] = 2 + return session + + +def test_clearing_a_session_forgets_the_turn_markers(): + """A marker says "the turn starting at index k was cut". After ``/new`` the + index starts again at 0, and a command that writes no marker of its own + (``/help``) begins at 0 <= k -- so ACP reported it cut.""" + session = _session_with_two_turns_the_last_one_cut() + + session.clear() + + assert [session.metadata[key] for key in _TURN_MARKERS] == [None, None] + + +def test_undoing_the_cut_turn_forgets_its_markers_and_keeps_an_earlier_ones(): + session = _session_with_two_turns_the_last_one_cut() + + session.undo_last_turn() + + assert [session.metadata[key] for key in _TURN_MARKERS] == [None, None] + + earlier = _session_with_two_turns_the_last_one_cut() + for key in _TURN_MARKERS: + earlier.metadata[key] = 0 + earlier.undo_last_turn() + assert [earlier.metadata[key] for key in _TURN_MARKERS] == [0, 0], "the turn it marks is still there" diff --git a/tests/test_subagent_acp.py b/tests/test_subagent_acp.py index 0e80a8620..e25275be1 100644 --- a/tests/test_subagent_acp.py +++ b/tests/test_subagent_acp.py @@ -2550,6 +2550,188 @@ async def test_dispatch_accepts_the_managers_full_keyword_set(tmp_path: Path) -> assert reply == "pong" +_TURN_COUNTS = {"inputTokens": 1200, "outputTokens": 34, "cachedReadTokens": 800, "totalTokens": 2034} + + +async def _counted(tmp_path: Path, **env: str) -> tuple[int | None, int | None]: + """The run record's token counts for one turn of a stub reporting usage as ``env`` says.""" + from raven.agent.subagent import activity + + cfg = stub_config("a") + cfg.env.update(env) + backend = build_third_party_backend(cfg) + with activity.collecting() as did: + await backend.run("ping", task_id="t1", workspace=tmp_path, executor=None) + return did.tokens_in, did.tokens_out + + +async def test_the_turns_token_counts_reach_the_run_record(tmp_path: Path) -> None: + """claude-agent-acp and codex-acp put a turn's counts on the prompt response. + + Their ``usage_update`` carries the window (``used``, ``size``) and the cost, + no counts, and only it was read -- so every ACP run's record said nothing + about tokens, and the tasks board showed none for an agent that sent them. + """ + assert await _counted(tmp_path, ACP_STUB_PROMPT_USAGE=json.dumps(_TURN_COUNTS)) == (1200, 34) + + +async def test_counts_in_both_places_are_counted_once(tmp_path: Path) -> None: + """``note_usage`` adds, so an agent that repeats its counts in the update + must not have the turn's tokens doubled.""" + both = { + "ACP_STUB_PROMPT_USAGE": json.dumps(_TURN_COUNTS), + "ACP_STUB_USAGE_UPDATE": json.dumps({"used": 42, "size": 1000, "inputTokens": 1200, "outputTokens": 34}), + } + assert await _counted(tmp_path, **both) == (1200, 34) + + +async def test_an_agent_counting_only_in_its_update_is_still_read(tmp_path: Path) -> None: + """The spelling an earlier agent was seen sending stays read.""" + update = {"used": 42, "size": 1000, "inputTokens": 1200, "outputTokens": 34} + assert await _counted(tmp_path, ACP_STUB_USAGE_UPDATE=json.dumps(update)) == (1200, 34) + + +@pytest.mark.parametrize( + ("agent", "version", "counted"), + [ + ("claude-agent-acp", "0.94.0", (1200, 34)), + ("codex-acp", "2.2.2", (1200, 34)), + ("codex-acp", "2.1.1", (None, None)), + ("codex-acp", "", (None, None)), + ], +) +async def test_prompt_usage_is_read_only_where_it_counts_the_turn( + tmp_path: Path, agent: str, version: str, counted: tuple[int | None, int | None] +) -> None: + """claude-agent-acp sums the turn's calls, and so does codex-acp from 2.2.1. + Before that codex-acp sent the last call alone (``tokenUsage.last``), so a + twenty-call turn would be recorded as one call's tokens -- a confident + undercount where there was no count before.""" + env = { + "ACP_STUB_AGENT_NAME": agent, + "ACP_STUB_AGENT_VERSION": version, + "ACP_STUB_PROMPT_USAGE": json.dumps(_TURN_COUNTS), + } + 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.""" + from datetime import datetime, timedelta, timezone + + from raven.token_wise import usage_context + from raven.token_wise.usage_tracker import delegated_usage + + cfg = stub_config("a") + cfg.env.update(env) + registry = InstanceRegistry(path=tmp_path / "instances.json") + if bound: + await registry.commit("cli:root", "a", "work", "earlier-session", kind="acp") + backend = AcpAgentBackend( + name="a", + command=cfg.command, + env=dict(cfg.env), + snapshot=_snapshot("a", cfg, can_resume=True, can_load=True), + registry=registry, + ) + 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( + "ping", task_id=f"t{n}", workspace=tmp_path, session_key="cli:root", instance="work", executor=None + ) + return delegated_usage("cli:root", since, telemetry_dir=ledger) + + +_CLAUDE_TURN = { + "ACP_STUB_AGENT_NAME": "claude-agent-acp", + "ACP_STUB_PROMPT_USAGE": json.dumps(_TURN_COUNTS), +} + + +async def test_a_third_party_agents_turn_is_booked_in_the_usage_ledger(tmp_path: Path) -> None: + """An in-process sub-agent's calls are booked under the conversation that + delegated them; an ACP agent's turn was not, so its spend was missing from + every total that conversation reported.""" + got = await _booked(tmp_path, ACP_STUB_COST_PER_PROMPT="0.0125", **_CLAUDE_TURN) + + assert (got.calls, got.input_tokens, got.output_tokens) == (1, 1200, 34) + assert got.cost_usd == pytest.approx(0.0125) + + +@pytest.mark.parametrize( + "reported", + [ + {}, + { + "ACP_STUB_AGENT_NAME": "codex-acp", + "ACP_STUB_AGENT_VERSION": "2.1.1", + "ACP_STUB_PROMPT_USAGE": json.dumps(_TURN_COUNTS), + }, + ], + ids=["no-counts", "last-call-counts"], +) +async def test_a_turn_whose_counts_are_unknown_still_books_its_cost(tmp_path: Path, reported: dict[str, str]) -> None: + """``usage_update`` carries the window and the cost and no counts, so a turn + can arrive with a cost and no counts that are the turn's. Its cost is booked + and its counts are left missing: not dropped with them, nor written as zeros.""" + got = await _booked(tmp_path, ACP_STUB_COST_PER_PROMPT="0.0125", **reported) + + assert got.calls == 1 + assert got.cost_usd == pytest.approx(0.0125) and got.cost_missing_calls == 0 + assert (got.input_missing_calls, got.output_missing_calls) == (1, 1) + + +async def test_a_turn_with_neither_counts_nor_cost_books_nothing(tmp_path: Path) -> None: + """A row that knows nothing would still count as a call.""" + got = await _booked(tmp_path) + + assert got.calls == 0 + + +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.""" + got = await _booked(tmp_path, 2, ACP_STUB_COST_PER_PROMPT="0.01", **_CLAUDE_TURN) + + assert got.calls == 2 + assert got.cost_usd == pytest.approx(0.02) + + +async def test_a_resumed_session_whose_earlier_spend_is_unknown_books_no_cost(tmp_path: Path) -> None: + """A session this process did not open (a restart, then a resume) reports a + total that includes turns booked before; its counts are booked, its cost is + left unknown rather than charged again.""" + got = await _booked(tmp_path, bound=True, ACP_STUB_COST_PER_PROMPT="0.5", **_CLAUDE_TURN) + + assert (got.calls, got.input_tokens) == (1, 1200) + assert got.cost_usd is None and got.cost_missing_calls == 1 + + +async def test_a_session_nobody_can_resume_leaves_no_spend_behind(tmp_path: Path) -> None: + """The cumulative cost is kept to measure a resumed session's next turn by. + A stateless agent's sessions are never resumed, and keeping one entry per + dispatch would grow for as long as the process lives.""" + cfg = stub_config("a") + cfg.env.update({**_CLAUDE_TURN, "ACP_STUB_COST_PER_PROMPT": "0.01"}) + backend = build_third_party_backend(cfg) + + await backend.run("ping", task_id="t1", workspace=tmp_path, session_key="cli:root", executor=None) + + assert backend.is_stateful is False + assert backend._session_spend == {} + + +async def test_a_raven_agent_is_not_booked_twice(tmp_path: Path) -> None: + """A raven agent books its own calls under the delegating conversation (the + prompt's ``raven.usage``); booking its turn here as well counts it twice.""" + env = {**_CLAUDE_TURN, "ACP_STUB_AGENT_NAME": "raven", "ACP_STUB_COST_PER_PROMPT": "0.0125"} + got = await _booked(tmp_path, **env) + + assert got.calls == 0 + + async def test_dispatch_truncates_to_max_output_chars(tmp_path: Path) -> None: backend = build_third_party_backend(stub_config("a", max_output_chars=2)) assert await backend.run("ping", task_id="t1", workspace=tmp_path, executor=None) == "po" @@ -2886,6 +3068,100 @@ async def test_a_pruned_session_falls_back_to_a_fresh_one(tmp_path: Path) -> Non assert rebound is not None and rebound != "pruned-session" +async def test_a_pruned_session_says_its_reply_starts_over(tmp_path: Path) -> None: + """The fresh session is a working outcome, and it is not the one asked for. + + The caller reused the handle to continue a conversation; the agent no + longer has it, so the reply comes from a session with none of that context + -- "apply the fix you proposed" answered from an empty history. Logged and + nothing else, the caller read it as the follow-up it asked for. + """ + cfg = stub_config("a") + registry = InstanceRegistry(path=tmp_path / "instances.json") + await registry.commit("s", "a", "work", "pruned-session", kind="acp") + backend = AcpAgentBackend( + name="a", + command=cfg.command, + env=dict(cfg.env), + snapshot=_snapshot("a", cfg, can_resume=True, can_load=True), + registry=registry, + ) + seen: list[str] = [] + + async def on_delta(text: str) -> None: + seen.append(text) + + reply = await backend.run( + "now apply it", + task_id="t1", + workspace=tmp_path, + session_key="s", + instance="work", + executor=None, + on_delta=on_delta, + ) + + note = ( + "\n\n[raven] Instance 'work' could not pick up its earlier conversation (Session not found), " + "so this reply comes from a fresh session that has none of it." + ) + assert reply.endswith(note) + assert seen[-1] == note, "a caller that streamed reads it too" + + +async def test_the_note_waits_for_a_reply_when_the_fresh_sessions_turn_fails(tmp_path: Path) -> None: + """The binding is dropped when the load is refused, so the next dispatch on the + handle opens yet another fresh session with no refusal of its own to report. + A note handed out only by the turn that hit the refusal is lost whenever that + turn fails -- and a caller retries precisely to continue the conversation.""" + from raven.acp_client.acp_agent import AcpEmptyTurnError + + cfg = stub_config("a") + cfg.env["ACP_STUB_EMPTY_PROMPTS"] = "1" + registry = InstanceRegistry(path=tmp_path / "instances.json") + await registry.commit("s", "a", "work", "pruned-session", kind="acp") + backend = AcpAgentBackend( + name="a", + command=cfg.command, + env=dict(cfg.env), + snapshot=_snapshot("a", cfg, can_resume=True, can_load=True), + registry=registry, + ) + + with pytest.raises(AcpEmptyTurnError): + await backend.run( + "now apply it", task_id="t1", workspace=tmp_path, session_key="s", instance="work", executor=None + ) + reply = await backend.run( + "now apply it", task_id="t2", workspace=tmp_path, session_key="s", instance="work", executor=None + ) + + assert "could not pick up its earlier conversation (Session not found)" in reply + again = await backend.run( + "and the tests", task_id="t3", workspace=tmp_path, session_key="s", instance="work", executor=None + ) + assert "[raven]" not in again, "said once, to the first reply" + + +async def test_a_resumed_session_carries_no_such_note(tmp_path: Path) -> None: + cfg = stub_config("a") + registry = InstanceRegistry(path=tmp_path / "instances.json") + await registry.commit("s", "a", "work", "live-session", kind="acp") + backend = AcpAgentBackend( + name="a", + command=cfg.command, + env=dict(cfg.env), + snapshot=_snapshot("a", cfg, can_resume=True, can_load=True), + registry=registry, + ) + + reply = await backend.run( + "go on", task_id="t1", workspace=tmp_path, session_key="s", instance="work", executor=None + ) + + assert "[raven]" not in reply + + async def test_load_fallback_reuses_the_same_mcp_grant(tmp_path: Path) -> None: cfg = stub_config("a") registry = InstanceRegistry(path=tmp_path / "instances.json") diff --git a/tests/test_subagent_github_copilot.py b/tests/test_subagent_github_copilot.py index 8e00dc3d3..f739eae4d 100644 --- a/tests/test_subagent_github_copilot.py +++ b/tests/test_subagent_github_copilot.py @@ -226,3 +226,135 @@ async def run(self, *args: object, **kwargs: object) -> str: result = await probe_mod.ping_agent(_cfg()) assert result.ok is True assert result.remedy is None + + +# ---- the dispatch path ------------------------------------------------------ + + +_STUB = Path(__file__).with_name("acp_stub_server.py") + + +@pytest.fixture +async def _closed_pool(): + from raven.acp_client.pool import close_pool + + yield + await close_pool() + + +def _dispatched(preset: str, reply: str): + import sys + + from raven.agent.subagent.backends import build_third_party_backend + from raven.config.schema import ThirdPartyAcpSubagentConfig + from raven.utils.commands import command_quote + + return build_third_party_backend( + ThirdPartyAcpSubagentConfig( + name="GitHub Copilot", + preset=preset, + command=f"{command_quote(sys.executable)} {command_quote(str(_STUB))}", + env={"ACP_STUB_REPLY": reply}, + ready_timeout_ms=15000, + ) + ) + + +async def test_a_dispatch_whose_reply_is_the_refusal_fails(tmp_path: Path, _closed_pool) -> None: + """The run a spawn or a DAG node makes, not only the Connect ping. + + Copilot ends the turn normally with the refusal as its message, so the + backend's own checks -- an empty turn, a failed call with nothing after it -- + pass it, and the sentence went back as the work: a node recorded completed + whose output was "Your credit balance is too low", merged downstream. + """ + from raven.acp_client.acp_agent import AcpEmptyTurnError + + backend = _dispatched("github_copilot", _CREDIT) + + with pytest.raises(AcpEmptyTurnError) as caught: + await backend.run("summarize the README", task_id="t1", workspace=tmp_path, executor=None) + + assert "for want of credit" in str(caught.value) + assert _CREDIT in str(caught.value), "the agent's own words travel with the reading" + + +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.""" + answer = ( + "The deploy failed because the worker crashed. Its log ends with " + "'Last error: 500 Internal server error'; restart the worker and redeploy." + ) + backend = _dispatched("github_copilot", answer) + + out = await backend.run("why did the deploy fail?", task_id="t2", workspace=tmp_path, executor=None) + + assert out == answer + + +@pytest.mark.parametrize("said", [_KEY_401, _KEY_403, _CREDIT, _EXPIRED, _MODEL, _LOCAL, _TUNNEL, _SERVER, _GATEWAY]) +def test_every_measured_refusal_is_read_as_a_whole_reply(said: str) -> None: + assert github_copilot.read_reply(said) is not None + + +@pytest.mark.parametrize( + "answer", + [ + "Error: 502 Bad Gateway from the staging load balancer; the upstream pool is empty.", + "Error: GET /api/users failed (HTTP 404) because the route was renamed to /api/accounts.", + "Error: your credit balance is too low is what the billing page shows; top it up there.", + ], +) +def test_an_answer_that_opens_with_an_error_it_found_is_not_a_refusal(answer: str) -> None: + """A task can be "curl staging and tell me what it returns": the answer then + opens with the error it found. Copilot's own refusals are its fixed sentences + -- "with provider at", "from the AI model; retried" -- and only those are.""" + assert github_copilot.read_reply(answer) is None + + +def test_a_long_reply_is_never_read_as_a_refusal() -> None: + """Copilot's refusals are short -- the longest measured runs to under 300 + characters. A long reply is work, whatever it opens with; and a long reply + that opens like one and never closes would walk the patterns above once per + restart, which measured 1.6 s at 0.56 MB and 23 s at 2.2 MB.""" + reply = "Error: Authentication failed with provider at http://x (HTTP 401)." + " and the rest of an answer" * 100 + + assert github_copilot.read_reply(reply) is None + + +async def test_another_agents_dispatch_is_not_read_with_copilots_sentences(tmp_path: Path, _closed_pool) -> None: + backend = _dispatched("qwen_code", _CREDIT) + + out = await backend.run("summarize the README", task_id="t3", workspace=tmp_path, executor=None) + + assert out == _CREDIT + + +async def test_a_connect_reads_a_refusal_once_now_that_the_run_raises_it( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _closed_pool +) -> None: + """The connect builds the same backend a dispatch does, so the refusal now + arrives as the run's failure rather than as its reply. Read again from the + failure's text, the reading wrapped itself: "refused ... It said: acp agent + ... refused ... It said: Error: 402 ...". It is read from the agent's words.""" + import sys + + from raven.agent.subagent import probe as probe_mod + from raven.config.schema import ThirdPartyAcpSubagentConfig + from raven.utils.commands import command_quote + + monkeypatch.chdir(tmp_path) + cfg = ThirdPartyAcpSubagentConfig( + name="GitHub Copilot", + preset="github_copilot", + command=f"{command_quote(sys.executable)} {command_quote(str(_STUB))}", + env={"ACP_STUB_REPLY": _CREDIT}, + ready_timeout_ms=15000, + ) + + result = await probe_mod.ping_agent(cfg) + + assert result.ok is False + assert result.remedy == Remedy("billing") + assert result.detail == github_copilot.read(_CREDIT)[0] diff --git a/tests/test_subagent_manager.py b/tests/test_subagent_manager.py index ef7c7b333..4f0aa52df 100644 --- a/tests/test_subagent_manager.py +++ b/tests/test_subagent_manager.py @@ -3306,11 +3306,55 @@ async def test_a_run_out_of_rounds_answers_from_what_it_gathered(tmp_path) -> No out = await backend.run("research it", task_id="n1", workspace=tmp_path, executor=None) - assert out == "here is what I have" + assert out.startswith("here is what I have\n\n[raven] ") assert provider.rounds == RavenLoopBackend._MAX_ITERATIONS, "the budget is still a budget" assert provider.toolless_calls == 1, "asked exactly once, after the rounds ran out" +async def test_a_run_out_of_rounds_says_its_answer_is_partial(tmp_path) -> None: + """The wrap-up is worth having and it is still not the work: the tool budget + ended the run, not the task. Handed back bare, a spawn records it completed + and a DAG node's judge reads it as the result. The ACP backend marks the same + case with a ``[raven]`` line (``AcpAgentBackend._finished``); so does this one, + in the reply and on the stream, which is the only place a watcher reads it. + """ + from raven.agent.subagent.backends.raven_loop import RavenLoopBackend + + backend = RavenLoopBackend(provider=_AlwaysToolsProvider(), model="stub", agent_home=tmp_path) + seen: list[str] = [] + + async def on_delta(text: str) -> None: + seen.append(text) + + out = await backend.run("research it", task_id="n8", workspace=tmp_path, executor=None, on_delta=on_delta) + + notice = ( + f"\n\n[raven] The sub-agent used all {RavenLoopBackend._MAX_ITERATIONS} tool rounds " + "before finishing; reply is partial." + ) + assert out == "here is what I have" + notice + assert seen == ["here is what I have", notice], seen + + +async def test_the_partial_notice_stays_out_of_the_instances_own_history(tmp_path) -> None: + """The notice is raven telling the caller about this run, not something the + sub-agent said. Kept in the history a resumed instance continues from, the + model reads it as its own earlier words and can repeat it in a reply that was + not cut -- which the caller would then read as cut. The ACP backend never puts + its notices into the agent's session, for the same reason.""" + from raven.agent.subagent.backends.raven_loop import RavenLoopBackend + + backend = RavenLoopBackend(provider=_AlwaysToolsProvider(), model="stub", agent_home=tmp_path) + kept: list[list[dict[str, Any]]] = [] + + out = await backend.run( + "research it", task_id="n10", workspace=tmp_path, executor=None, on_messages=lambda m: kept.append(list(m)) + ) + + assert "reply is partial" in out + assert kept[-1][-1] == {"role": "assistant", "content": "here is what I have"} + + async def test_a_run_with_nothing_to_say_fails_instead_of_reading_as_done(tmp_path) -> None: """Raised, not returned. A node that produced nothing must not wear a tick while the step downstream merges its placeholder as data.""" @@ -3456,8 +3500,8 @@ async def on_delta(text: str) -> None: out = await backend.run("research it", task_id="n7", workspace=tmp_path, executor=None, on_delta=on_delta) - assert out == "here is what I have" - assert seen == ["here is what I have"] + assert seen[0] == "here is what I have" + assert "".join(seen) == out async def test_a_wrap_up_that_answers_nothing_streams_nothing(tmp_path) -> None: diff --git a/tests/test_token_wise_usage_tracker.py b/tests/test_token_wise_usage_tracker.py index b4249e786..dc4d60f17 100644 --- a/tests/test_token_wise_usage_tracker.py +++ b/tests/test_token_wise_usage_tracker.py @@ -450,6 +450,24 @@ async def test_in_process_subagent_usage_is_written_as_delegated_usage(tmp_path: assert got.cost_usd == pytest.approx(0.2) +async def test_a_delegated_call_without_counts_is_booked_with_its_cost_alone(tmp_path: Path): + """Its counts are stored as missing, not as the zeros ``normalize_usage`` + fills in: a zero says the call read and wrote nothing.""" + from datetime import datetime, timedelta, timezone + + from raven.token_wise.usage_tracker import delegated_usage, record_delegated_usage + + since = datetime.now(timezone.utc) - timedelta(seconds=1) + with usage_context.bind("cli:root", {"root_session_key": "cli:root", "telemetry_dir": str(tmp_path)}): + await record_delegated_usage({"cost_usd": 0.0125}, model="m", root="cli:root", task_id="t1", counts_known=False) + + (row,) = [json.loads(line) for path in tmp_path.glob("usage-*.jsonl") for line in path.read_text().splitlines()] + assert (row["input_tokens"], row["output_tokens"], row["cost_usd"]) == (None, None, 0.0125) + got = delegated_usage("cli:root", since, telemetry_dir=tmp_path) + assert (got.calls, got.input_missing_calls, got.output_missing_calls) == (1, 1, 1) + assert got.cost_usd == pytest.approx(0.0125) + + def test_delegated_usage_keeps_to_its_window_and_to_reported_prices(tmp_path: Path): from datetime import datetime, timedelta, timezone