Surface runtime model + token usage in published review summaries

src/code_review_bot/agent/acp.py

@@ -227,6 +227,7 @@ async def request_permission(
             # in _meta which the server ignores for model selection.
             env = {**(env or {}), "ANTHROPIC_MODEL": config.model}
         usage_dict: dict[str, Any] = {}
+        response_model: str | None = None
         async with spawn_agent_process(
             client,
             config.command,
@@ -272,8 +273,13 @@ async def request_permission(
                     session_id=session.session_id,
                     prompt=[text_block(combined_prompt)],
                 )
-                if response is not None and response.usage is not None:
-                    usage_dict = _usage_to_dict(response.usage)
+                if response is not None:
+                    raw_model = getattr(response, "model", None)
+                    response_model = (
+                        raw_model if isinstance(raw_model, str) and raw_model.strip() else None
+                    )
+                    if response.usage is not None:
+                        usage_dict = _usage_to_dict(response.usage)
             except Exception:
                 _log_acp_error(
                     _proc,
@@ -290,7 +296,7 @@ async def request_permission(
         _log_usage(usage_dict)
         text = client.collected_text.strip()
         parts: list[dict[str, Any]] = [{"type": "text", "text": text}] if text else []
-        return AgentRunResult(text=text, parts=parts, usage=usage_dict)
+        return AgentRunResult(text=text, parts=parts, usage=usage_dict, model=response_model)
 
 
 def _log_acp_error(proc: object, **context: object) -> None:

src/code_review_bot/agent/protocol.py

@@ -9,6 +9,7 @@ class AgentRunResult:
     text: str
     parts: list[dict[str, Any]]
     usage: dict[str, Any]
+    model: str | None = None
 
 
 class CodingAgent(Protocol):

src/code_review_bot/review/publish/debug.py

@@ -47,6 +47,7 @@ async def publish(
             skill_name=skill_name,
             skill_version=skill_version,
             fingerprints=fingerprints,
+            runtime=result.runtime,
         )
         summary_note_clean = _strip_metadata_comment(summary_note)
 

src/code_review_bot/review/publish/formatter.py

@@ -3,7 +3,7 @@
 import json
 
 from code_review_bot.platforms.models import ChangeRequest
-from code_review_bot.skill.protocol import Finding, count_findings_by_severity
+from code_review_bot.skill.protocol import Finding, RuntimeMetadata, count_findings_by_severity
 
 BOT_METADATA_PREFIX = "<!-- code-review-bot:"
 CODE_REVIEW_BOT_URL = "https://github.com/whhe/code-review-bot"
@@ -26,6 +26,7 @@ def format_review_note(
     located_count: int = 0,
     unlocated_findings: list[Finding] | None = None,
     resolved_findings: list[Finding] | None = None,
+    runtime: RuntimeMetadata | None = None,
 ) -> str:
     findings = unlocated_findings or []
     counts = severity_counts or count_findings_by_severity(findings)
@@ -67,6 +68,9 @@ def format_review_note(
                 f"{finding.description}~~"
             )
         lines.append("")
+    if lines and lines[-1] != "":
+        lines.append("")
+    lines.append(_format_runtime_line(runtime))
     lines.append(
         f"_Generated by [whhe/code-review-bot]({CODE_REVIEW_BOT_URL}) · "
         f"Skill fingerprint: `{skill_version}`_"
@@ -79,3 +83,20 @@ def format_review_note(
     }
     lines.append(f"{BOT_METADATA_PREFIX}{json.dumps(metadata, separators=(',', ':'))} -->")
     return "\n".join(lines)
+
+
+def _format_runtime_line(runtime: RuntimeMetadata | None) -> str:
+    model = runtime.model if runtime and runtime.model else "unavailable"
+    input_tokens = _format_token_value(runtime.input_tokens if runtime else None)
+    output_tokens = _format_token_value(runtime.output_tokens if runtime else None)
+    total_tokens = _format_token_value(runtime.total_tokens if runtime else None)
+    return (
+        f"Model: {model} · Tokens: input {input_tokens} / "
+        f"output {output_tokens} / total {total_tokens}"
+    )
+
+
+def _format_token_value(value: int | None) -> str:
+    if value is None:
+        return "unavailable"
+    return f"{value:,}"

src/code_review_bot/review/publish/platform.py

@@ -45,6 +45,7 @@ async def publish(
             skill_name=skill_name,
             skill_version=skill_version,
             fingerprints=fingerprints,
+            runtime=result.runtime,
         )
         if publish_summary:
             await self.adapter.publish_summary(cr.project_ref, cr.cr_id, body)

src/code_review_bot/review/runner.py

@@ -3,7 +3,7 @@
 from code_review_bot.agent.json_extract import complete_json_with_retries
 from code_review_bot.agent.protocol import CodingAgent
 from code_review_bot.review.models import ReviewTaskContext
-from code_review_bot.skill.protocol import SkillResult
+from code_review_bot.skill.protocol import RuntimeMetadata, SkillResult
 
 
 class PromptBuildingSkill(Protocol):
@@ -22,18 +22,70 @@ def __init__(self, agent: CodingAgent, max_json_retries: int = 1) -> None:
 
     async def review(self, skill: PromptBuildingSkill, context: ReviewTaskContext) -> SkillResult:
         prompt = skill.build_prompt(context)
+        model_candidates: list[str] = []
+        token_keys = ("input_tokens", "output_tokens", "total_tokens")
+        token_totals = {key: 0 for key in token_keys}
+        token_valid_counts = {key: 0 for key in token_keys}
+        run_count = 0
+
+        def collect_runtime_metadata(result: object) -> None:
+            nonlocal run_count
+            run_count += 1
+            model = getattr(result, "model", None)
+            if isinstance(model, str) and model.strip():
+                model_candidates.append(model.strip())
+            usage = getattr(result, "usage", {})
+            usage_map = usage if isinstance(usage, dict) else {}
+            for key in token_keys:
+                value = usage_map.get(key)
+                if isinstance(value, int) and not isinstance(value, bool):
+                    token_totals[key] += value
+                    token_valid_counts[key] += 1
 
         async def run(current_prompt: str) -> str:
             result = await self.agent.run_once(
                 current_prompt,
                 agent="plan",
                 additional_directories=getattr(skill, "additional_directories", []),
             )
+            collect_runtime_metadata(result)
             return result.text
 
-        return await complete_json_with_retries(
+        parsed = await complete_json_with_retries(
             prompt,
             SkillResult,
             run,
             max_retries=self.max_json_retries,
         )
+        model_value: str | None = None
+        unique_models = list(dict.fromkeys(model_candidates))
+        if len(unique_models) == 1:
+            model_value = unique_models[0]
+        elif len(unique_models) > 1:
+            model_value = f"multiple models used: {', '.join(unique_models)}"
+
+        def aggregate_tokens(key: str) -> int | None:
+            # Show totals only when every call reported the metric so the final
+            # summary never presents partial usage as complete usage. A partial
+            # sum can under-report total tokens and look authoritative.
+            if token_valid_counts[key] != run_count:
+                return None
+            return token_totals[key]
+
+        runtime = RuntimeMetadata(
+            model=model_value,
+            input_tokens=aggregate_tokens("input_tokens"),
+            output_tokens=aggregate_tokens("output_tokens"),
+            total_tokens=aggregate_tokens("total_tokens"),
+        )
+        if not any(
+            value is not None
+            for value in (
+                runtime.model,
+                runtime.input_tokens,
+                runtime.output_tokens,
+                runtime.total_tokens,
+            )
+        ):
+            return parsed
+        return parsed.with_runtime(runtime)

src/code_review_bot/skill/protocol.py

@@ -1,6 +1,6 @@
 from typing import Any, Literal, Protocol
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, PrivateAttr, field_validator
 
 _SEVERITIES = ("critical", "high", "medium", "low")
 
@@ -77,6 +77,23 @@ class SkillResult(BaseModel):
 
     summary: str
     findings: list[Finding] = Field(default_factory=list)
+    _runtime: "RuntimeMetadata | None" = PrivateAttr(default=None)
+
+    @property
+    def runtime(self) -> "RuntimeMetadata | None":
+        return self._runtime
+
+    def with_runtime(self, runtime: "RuntimeMetadata | None") -> "SkillResult":
+        copied = self.model_copy(deep=True)
+        copied._runtime = runtime
+        return copied
+
+
+class RuntimeMetadata(BaseModel):
+    model: str | None = None
+    input_tokens: int | None = None
+    output_tokens: int | None = None
+    total_tokens: int | None = None
 
 
 class ReviewSkill(Protocol):

tests/test_acp_runtime.py

@@ -85,3 +85,33 @@ async def fake_spawn_agent_process(
     assert connection.model_calls == expected_model_calls
     assert connection.config_calls == expected_config_calls
     assert captured_env.get("ANTHROPIC_MODEL") == expected_env_model
+
+
+@pytest.mark.asyncio
+async def test_acp_runtime_returns_runtime_model_from_agent_response(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    connection = _Connection()
+
+    async def prompt_with_model(**kwargs: object) -> SimpleNamespace:
+        return SimpleNamespace(usage=None, model="provider/runtime-model")
+
+    connection.prompt = prompt_with_model  # type: ignore[method-assign]
+
+    @asynccontextmanager
+    async def fake_spawn_agent_process(
+        client: object,
+        command: str,
+        *args: str,
+        env: dict[str, str] | None = None,
+        **kwargs: object,
+    ) -> AsyncIterator[tuple[_Connection, SimpleNamespace]]:
+        yield connection, SimpleNamespace(stdin=None, stdout=None, stderr=None, _transport=None)
+
+    monkeypatch.setattr(acp_sdk, "spawn_agent_process", fake_spawn_agent_process)
+    agent = AcpCodingAgent(AcpAgentConfig(command="agent", verbose=False), tmp_path)
+
+    run = await agent.run_once("review")
+
+    assert run.model == "provider/runtime-model"

tests/test_review_publisher.py

@@ -4,7 +4,7 @@
 from code_review_bot.review.publish.debug import DebugMarkdownPublisher
 from code_review_bot.review.publish.formatter import BOT_METADATA_PREFIX, format_review_note
 from code_review_bot.review.publish.platform import PlatformPublisher
-from code_review_bot.skill.protocol import Finding, SkillResult
+from code_review_bot.skill.protocol import Finding, RuntimeMetadata, SkillResult
 
 
 def make_change_request() -> ChangeRequest:
@@ -38,6 +38,17 @@ def make_finding(**overrides: object) -> Finding:
     return Finding(**data)
 
 
+def make_runtime(**overrides: object) -> RuntimeMetadata:
+    data: dict[str, object] = {
+        "model": "provider/model",
+        "input_tokens": 12345,
+        "output_tokens": 678,
+        "total_tokens": 13023,
+    }
+    data.update(overrides)
+    return RuntimeMetadata(**data)
+
+
 class FakeAdapter:
     """In-memory PlatformAdapter for tests."""
 
@@ -182,6 +193,62 @@ async def test_formatter_includes_unlocated_finding() -> None:
     assert "A risky pattern" in body
 
 
+@pytest.mark.asyncio
+async def test_formatter_includes_runtime_line() -> None:
+    body = format_review_note(
+        cr=make_change_request(),
+        summary="Reviewed",
+        severity_counts={"critical": 0, "high": 1, "medium": 0, "low": 0},
+        located_count=1,
+        unlocated_findings=[],
+        skill_name="default",
+        skill_version="1",
+        fingerprints=["fp1"],
+        runtime=make_runtime(),
+    )
+
+    assert "Model: provider/model · Tokens: input 12,345 / output 678 / total 13,023" in body
+
+
+@pytest.mark.asyncio
+async def test_formatter_marks_unavailable_runtime_values() -> None:
+    body = format_review_note(
+        cr=make_change_request(),
+        summary="Reviewed",
+        severity_counts={"critical": 0, "high": 1, "medium": 0, "low": 0},
+        located_count=1,
+        unlocated_findings=[],
+        skill_name="default",
+        skill_version="1",
+        fingerprints=["fp1"],
+        runtime=make_runtime(model=None, output_tokens=None),
+    )
+
+    assert (
+        "Model: unavailable · Tokens: input 12,345 / output unavailable / total 13,023" in body
+    )
+
+
+@pytest.mark.asyncio
+async def test_formatter_marks_unavailable_when_runtime_missing() -> None:
+    body = format_review_note(
+        cr=make_change_request(),
+        summary="Reviewed",
+        severity_counts={"critical": 0, "high": 1, "medium": 0, "low": 0},
+        located_count=1,
+        unlocated_findings=[],
+        skill_name="default",
+        skill_version="1",
+        fingerprints=["fp1"],
+        runtime=None,
+    )
+
+    assert (
+        "Model: unavailable · Tokens: input unavailable / output unavailable / total unavailable"
+        in body
+    )
+
+
 @pytest.mark.asyncio
 async def test_publisher_renders_resolved_findings() -> None:
     adapter = FakeAdapter()