ACP adapter missing usage field population fix

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #7086 fix(acp): populate PromptResponse usage from top-level result fields

  merged to main

  files: acp_adapter/server.py, tests/acp/test_server.py

  Show the diff

  • acp_adapter/server.py

    @@ -451,14 +451,13 @@ def _run_agent() -> dict:
                 await conn.session_update(session_id, update)
     
             usage = None
    -        usage_data = result.get("usage")
    -        if usage_data and isinstance(usage_data, dict):
    +        if any(result.get(key) is not None for key in ("prompt_tokens", "completion_tokens", "total_tokens")):
                 usage = Usage(
    -                input_tokens=usage_data.get("prompt_tokens", 0),
    -                output_tokens=usage_data.get("completion_tokens", 0),
    -                total_tokens=usage_data.get("total_tokens", 0),
    -                thought_tokens=usage_data.get("reasoning_tokens"),
    -                cached_read_tokens=usage_data.get("cached_tokens"),
    +                input_tokens=result.get("prompt_tokens", 0),
    +                output_tokens=result.get("completion_tokens", 0),
    +                total_tokens=result.get("total_tokens", 0),
    +                thought_tokens=result.get("reasoning_tokens"),
    +                cached_read_tokens=result.get("cache_read_tokens"),
                 )
     
             stop_reason = "cancelled" if state.cancel_event and state.cancel_event.is_set() else "end_turn"

  • tests/acp/test_server.py

    @@ -410,6 +410,37 @@ async def test_prompt_sends_final_message_update(self, agent):
             update = last_call[1].get("update") or last_call[0][1]
             assert update.session_update == "agent_message_chunk"
     
    +    @pytest.mark.asyncio
    +    async def test_prompt_populates_usage_from_top_level_run_conversation_fields(self, agent):
    +        """ACP should map top-level token fields into PromptResponse.usage."""
    +        new_resp = await agent.new_session(cwd=".")
    +        state = agent.session_manager.get_session(new_resp.session_id)
    +
    +        state.agent.run_conversation = MagicMock(return_value={
    +            "final_response": "usage attached",
    +            "messages": [],
    +            "prompt_tokens": 123,
    +            "completion_tokens": 45,
    +            "total_tokens": 168,
    +            "reasoning_tokens": 7,
    +            "cache_read_tokens": 11,
    +        })
    +
    +        mock_conn = MagicMock(spec=acp.Client)
    +        mock_conn.session_update = AsyncMock()
    +        agent._conn = mock_conn
    +
    +        prompt = [TextContentBlock(type="text", text="show usage")]
    +        resp = await agent.prompt(prompt=prompt, session_id=new_resp.session_id)
    +
    +        assert isinstance(resp, PromptResponse)
    +        assert resp.usage is not None
    +        assert resp.usage.input_tokens == 123
    +        assert resp.usage.output_tokens == 45
    +        assert resp.usage.total_tokens == 168
    +        assert resp.usage.thought_tokens == 7
    +        assert resp.usage.cached_read_tokens == 11
    +
         @pytest.mark.asyncio
         async def test_prompt_cancelled_returns_cancelled_stop_reason(self, agent):
             """If cancel is called during prompt, stop_reason should be 'cancelled'."""