Fix stack 07: tool-audit and tool-event parity

src/agent_runtime_kit/adapters/antigravity.py

@@ -493,14 +493,17 @@ async def _consume_chunk(
             )
             return
         if isinstance(chunk, sdk.types.ToolCall):
+            name = _tool_name(getattr(chunk, "name", "tool"))
+            arguments = _tool_arguments(chunk)
+            # Record the requested call now so result.tool_calls counts every
+            # invocation (matching Claude/Codex), including ones that never emit a
+            # ToolResult chunk. The matching result fills it in below.
+            tool_calls.append(
+                ToolCallAudit(tool_name=name, arguments=arguments, status="requested")
+            )
             await safe_emit(
                 task,
-                tool_requested_event(
-                    task,
-                    self.kind,
-                    tool_name=_tool_name(getattr(chunk, "name", "tool")),
-                    arguments=_tool_arguments(chunk),
-                ),
+                tool_requested_event(task, self.kind, tool_name=name, arguments=arguments),
             )
             return
         if isinstance(chunk, sdk.types.ToolResult):
@@ -510,7 +513,7 @@ async def _consume_chunk(
                 result_preview=str(getattr(chunk, "result", ""))[:256],
                 status=_tool_result_status(chunk),
             )
-            tool_calls.append(audit)
+            _attach_tool_result(tool_calls, audit)
             await safe_emit(task, tool_completed_event(task, self.kind, audit))
             return
         await safe_emit(
@@ -807,6 +810,32 @@ def _builtin_tool_values(builtin: Any) -> set[str] | None:
     return {str(getattr(member, "value", member)) for member in members}
 
 
+def _attach_tool_result(tool_calls: list[ToolCallAudit], result_audit: ToolCallAudit) -> None:
+    """Fill the most recent matching requested call with its result, else append.
+
+    Keeps result.tool_calls one-entry-per-invocation: a ToolCall recorded the
+    pending audit; its ToolResult replaces it here rather than adding a duplicate.
+    Vendor chunks carry no correlation id, so matching is by tool name against
+    the most recent pending call — exact for the sequential call/result pattern
+    the SDK emits, best-effort if same-name calls ever interleave.
+    """
+
+    for index in range(len(tool_calls) - 1, -1, -1):
+        existing = tool_calls[index]
+        if existing.status == "requested" and existing.tool_name == result_audit.tool_name:
+            tool_calls[index] = ToolCallAudit(
+                tool_name=result_audit.tool_name,
+                # ToolResult chunks may omit args; keep the request-time arguments
+                # rather than degrading the audit to an empty mapping.
+                arguments=result_audit.arguments or existing.arguments,
+                result_preview=result_audit.result_preview,
+                status=result_audit.status,
+                duration_ms=result_audit.duration_ms,
+            )
+            return
+    tool_calls.append(result_audit)
+
+
 def _tool_result_status(chunk: Any) -> str:
     if getattr(chunk, "error", None) is not None:
         return "error"

src/agent_runtime_kit/adapters/claude.py

@@ -155,6 +155,7 @@ async def run(self, task: AgentTask) -> AgentResult:
                 model=model,
                 dropped_options=dropped,
                 tool_results=stream.tool_results,
+                tool_previews=stream.tool_previews,
                 permission_mode=_effective_permission_mode(task.permissions),
                 process_metadata=(
                     self._process_reuse_metadata(process_reused)
@@ -369,6 +370,11 @@ def __init__(self, runtime: ClaudeAgentRuntime, task: AgentTask) -> None:
         self._tool_names: dict[str, str] = {}
         # tool_use_id -> "ok" | "error", populated from ToolResultBlock messages.
         self.tool_results: dict[str, str] = {}
+        # tool_use_id -> result preview, so result.tool_calls carry a capped raw
+        # preview of each tool result (previously result audits had none). Only
+        # the RESULT carries content; the sanitized tool_completed events expose
+        # just result_preview_length, never the preview text itself.
+        self.tool_previews: dict[str, str] = {}
 
     async def consume(self, message: Any) -> None:
         self.messages.append(message)
@@ -400,11 +406,13 @@ async def consume(self, message: Any) -> None:
                     continue
                 tool_use_id = optional_str(field_value(block, "tool_use_id"))
                 status = "error" if field_value(block, "is_error", False) else "ok"
+                preview = str(field_value(block, "content", ""))[:256]
                 if tool_use_id is not None:
                     self.tool_results[tool_use_id] = status
+                    self.tool_previews[tool_use_id] = preview
                 audit = ToolCallAudit(
                     tool_name=self._tool_names.get(tool_use_id or "", "tool"),
-                    result_preview=str(field_value(block, "content", ""))[:256],
+                    result_preview=preview,
                     status=status,
                 )
                 await safe_emit(task, tool_completed_event(task, kind, audit))
@@ -437,6 +445,7 @@ def _translate_messages(
     model: str,
     dropped_options: list[str],
     tool_results: Mapping[str, str],
+    tool_previews: Mapping[str, str],
     permission_mode: str,
     process_metadata: Mapping[str, Any] | None = None,
 ) -> AgentResult:
@@ -496,7 +505,7 @@ def _translate_messages(
         total_tokens=usage.total_tokens,
         cost_usd=cost_usd,
     )
-    tool_calls = _apply_tool_results(tool_calls, tool_use_ids, tool_results)
+    tool_calls = _apply_tool_results(tool_calls, tool_use_ids, tool_results, tool_previews)
     metadata: dict[str, Any] = {
         "model": model,
         "sdk": "claude_agent_sdk",
@@ -534,19 +543,23 @@ def _apply_tool_results(
     tool_calls: list[ToolCallAudit],
     tool_use_ids: list[str | None],
     tool_results: Mapping[str, str],
+    tool_previews: Mapping[str, str],
 ) -> list[ToolCallAudit]:
-    if not tool_results:
+    if not tool_results and not tool_previews:
         return tool_calls
     updated: list[ToolCallAudit] = []
     for audit, tool_use_id in zip(tool_calls, tool_use_ids, strict=True):
-        status = tool_results.get(tool_use_id or "")
-        if status is not None and status != audit.status:
+        key = tool_use_id or ""
+        status = tool_results.get(key)
+        preview = tool_previews.get(key, audit.result_preview)
+        new_status = status if status is not None else audit.status
+        if new_status != audit.status or preview != audit.result_preview:
             updated.append(
                 ToolCallAudit(
                     tool_name=audit.tool_name,
                     arguments=audit.arguments,
-                    result_preview=audit.result_preview,
-                    status=status,
+                    result_preview=preview,
+                    status=new_status,
                     duration_ms=audit.duration_ms,
                 )
             )

src/agent_runtime_kit/adapters/codex.py

@@ -39,6 +39,8 @@
     task_completed_event,
     task_failed_event,
     task_started_event,
+    tool_completed_event,
+    tool_requested_event,
 )
 
 # Vendor ``ThreadItem`` discriminator values that carry a tool/command invocation.
@@ -159,6 +161,19 @@ async def run(self, task: AgentTask) -> AgentResult:
             await safe_emit(task, task_failed_event(task, self.kind, error=str(exc)))
             raise
 
+        # Codex is non-streaming, so tool calls are parsed from the final TurnResult.
+        # Emit them as events anyway so an observability sink sees the same tool
+        # activity here as it does from the streaming Claude/Antigravity adapters,
+        # instead of only in result.tool_calls.
+        for audit in result.tool_calls:
+            await safe_emit(
+                task,
+                tool_requested_event(
+                    task, self.kind, tool_name=audit.tool_name, arguments=audit.arguments
+                ),
+            )
+            await safe_emit(task, tool_completed_event(task, self.kind, audit))
+
         if result.output:
             await safe_emit(task, output_delta_event(task, self.kind, text=result.output))
         if result.error:

tests/test_antigravity_adapter.py

@@ -206,14 +206,58 @@ async def test_antigravity_runtime_runs_with_injected_sdk(tmp_path: Path) -> Non
     assert result.output == "done: task"
     assert result.parsed_output == {"ok": True}
     assert result.session_id == "ag-session"
+    # ToolCall + its ToolResult must collapse into ONE completed audit entry,
+    # not a requested/ok pair — this is the cardinality contract.
+    assert len(result.tool_calls) == 1
     assert result.tool_calls[0].tool_name == "Read"
+    assert result.tool_calls[0].status == "ok"
     assert result.usage.input_tokens == 4
     assert sink.events[-1]["name"] == "agent.task.completed"
     assert FakeAgent.last_config is not None
     assert FakeAgent.last_config.kwargs["workspaces"] == [str(tmp_path)]
     assert FakeAgent.last_config.kwargs["api_key"] == "key"
 
 
+@pytest.mark.asyncio
+async def test_antigravity_counts_requested_tool_call_without_result(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    async def only_call(prompt: str):
+        yield FakeTypes.Text("working")
+        yield FakeTypes.ToolCall("Search", {"q": "x"})
+
+    monkeypatch.setattr(FakeAgent, "chunks_factory", staticmethod(only_call))
+    runtime = make_runtime(data_dir=tmp_path)
+
+    result = await runtime.run(AgentTask(goal="x"))
+
+    # A tool call with no ToolResult chunk is still counted (cardinality parity).
+    assert len(result.tool_calls) == 1
+    assert result.tool_calls[0].tool_name == "Search"
+    assert result.tool_calls[0].status == "requested"
+
+
+@pytest.mark.asyncio
+async def test_antigravity_tool_result_without_args_keeps_requested_arguments(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    async def argless_result(prompt: str):
+        yield FakeTypes.ToolCall("Read", {"path": "README.md"})
+        # ToolResult chunks may omit args entirely.
+        yield FakeTypes.ToolResult("Read", "contents")
+
+    monkeypatch.setattr(FakeAgent, "chunks_factory", staticmethod(argless_result))
+    runtime = make_runtime(data_dir=tmp_path)
+
+    result = await runtime.run(AgentTask(goal="x"))
+
+    assert len(result.tool_calls) == 1
+    assert result.tool_calls[0].status == "ok"
+    # The merged audit keeps the request-time arguments instead of degrading
+    # to an empty mapping when the result chunk carries none.
+    assert result.tool_calls[0].arguments == {"path": "README.md"}
+
+
 @pytest.mark.asyncio
 async def test_antigravity_session_id_falls_back_to_task(tmp_path: Path) -> None:
     class NoIdResponse:

tests/test_claude_adapter.py

@@ -578,6 +578,34 @@ async def test_claude_max_turns_finish_reason() -> None:
     assert result.error is not None
 
 
+@pytest.mark.asyncio
+async def test_claude_result_tool_calls_carry_result_preview() -> None:
+    messages = [
+        {
+            "type": "AssistantMessage",
+            "content": [
+                {"type": "text", "text": "run"},
+                {"type": "tool_use", "id": "t1", "name": "Read", "input": {"path": "x"}},
+            ],
+        },
+        {
+            "type": "UserMessage",
+            "content": [
+                {"type": "tool_result", "tool_use_id": "t1", "content": "file body"},
+            ],
+        },
+        result_message(),
+    ]
+    runtime = ClaudeAgentRuntime(query_func=make_query(messages), options_cls=FakeClaudeOptions)
+
+    result = await runtime.run(AgentTask(goal="x"))
+
+    # result.tool_calls now carry the same preview the streamed events do.
+    assert result.tool_calls[0].tool_name == "Read"
+    assert result.tool_calls[0].result_preview == "file body"
+    assert result.tool_calls[0].status == "ok"
+
+
 @pytest.mark.asyncio
 async def test_claude_unsatisfied_output_schema_fails() -> None:
     # Schema requested, no native structured output, text is not JSON -> failure,

tests/test_codex_adapter.py

@@ -474,6 +474,28 @@ async def test_codex_sandbox_mapping(filesystem: FilesystemAccess, expected: str
     assert FakeThread.last_run_kwargs["sandbox"] == expected
 
 
+@pytest.mark.asyncio
+async def test_codex_emits_tool_events() -> None:
+    run_result = {
+        "status": "completed",
+        "final_response": "done",
+        "items": [
+            {"type": "commandExecution", "command": "ls", "aggregated_output": "ok"},
+        ],
+    }
+    sink = RecordingEventSink()
+    runtime = make_runtime(run_result)
+
+    result = await runtime.run(AgentTask(goal="x", event_sink=sink))
+
+    names = [event["name"] for event in sink.events]
+    # Codex now emits tool events (parsed from the TurnResult) like the streaming
+    # adapters, not just result.tool_calls.
+    assert "agent.tool.requested" in names
+    assert "agent.tool.completed" in names
+    assert result.tool_calls[0].tool_name == "command"
+
+
 @pytest.mark.asyncio
 async def test_codex_usage_prefers_per_turn_over_cumulative() -> None:
     # A resumed thread reports cumulative 'total'; the per-turn 'last' is what this