feat: Report Claude SDK and Codex tool, approval and compaction activity

src/band/adapters/claude_sdk.py

@@ -131,6 +131,7 @@
     BandMCPTransport,
     SharedBandMCPBackend,
 )
+from band.runtime.activity import set_permission_wait
 from band.runtime.custom_tools import (
     CustomToolDef,
     custom_tool_effects,
@@ -142,6 +143,7 @@
 )
 from band.runtime.formatters import format_tokens, strip_leading_mentions
 from band.runtime.prompts import SYSTEM_UPDATE_PREFIX
+from band.runtime.toolactivity import end_tool_activity, report_tool_activity
 from band.runtime.tools import (
     ALL_TOOL_NAMES,
     BAND_MCP_SERVER_NAME,
@@ -469,6 +471,12 @@ class ClaudeSDKCommand(StrEnum):
 # Approvals only gate Claude Code's side-effecting native tools.
 NATIVE_TOOL_MATCHER = rf"^(?!{re.escape(MCP_TOOL_PREFIX)}|{re.escape(TOOL_SEARCH)}$).+"
 
+
+def _served_by_band(sdk_tool_name: str) -> bool:
+    """Band's MCP server reports its own tools' activity as they run."""
+    return sdk_tool_name.startswith(MCP_TOOL_PREFIX)
+
+
 # Patterns that look like secrets/tokens in shell commands
 _REDACT_RE = re.compile(
     r"""(?x)
@@ -675,7 +683,7 @@ def __init__(
         # _declined_the_reply for what this is cross-referenced against and
         # why tool_use_id uniqueness needs no turn-scoping here.
         self._notified_declines: dict[str, set[str]] = {}
-        # Bare tool name per pending call, keyed by room then tool_use_id
+        # SDK tool name per pending call, keyed by room then tool_use_id
         # (ToolResultBlock only carries the id). Room-scoped rather than
         # turn-local: a resumed session can replay a result whose tool_use
         # streamed in an earlier, truncated turn, and that result must still
@@ -1405,7 +1413,9 @@ async def _on_tool_use(
         # Bare name for the cross-adapter tool_call record (the SDK
         # namespaces our tools mcp__band__*; see _semantic_tool_name).
         tool_name = self._semantic_tool_name(block.name)
-        pending_tool_names[block.id] = tool_name
+        pending_tool_names[block.id] = block.name
+        if not _served_by_band(block.name):
+            await report_tool_activity(tools, tool_name)
         await self._narrate_tool_call(block, tool_name, room_id, tools)
 
     async def _narrate_tool_call(
@@ -1555,7 +1565,12 @@ async def _on_tool_result(
         Shared by both envelopes a result can arrive in: user-type messages
         (the protocol shape) and assistant messages (accepted defensively).
         """
-        result_tool_name = pending_tool_names.pop(block.tool_use_id, None)
+        sdk_tool_name = pending_tool_names.pop(block.tool_use_id, None)
+        if sdk_tool_name is not None and not _served_by_band(sdk_tool_name):
+            await end_tool_activity(tools)
+        result_tool_name = (
+            None if sdk_tool_name is None else self._semantic_tool_name(sdk_tool_name)
+        )
         await self._narrate_tool_result(block, result_tool_name, room_id, tools)
 
     async def _narrate_tool_result(
@@ -1858,36 +1873,41 @@ async def _resolve_manual_approval(
         # so the caller isn't left waiting for a timeout nobody will see.
         tools = self._room_tools.get(room_id)
         mention = [requester["id"]] if requester else None
-        if tools:
-            try:
-                await tools.send_notice(
-                    APPROVAL_REQUESTED_TEMPLATE.format(summary=summary, token=token),
-                    mentions=mention,
-                )
-            except Exception:  # noqa: BLE001 -- tool calls may raise any exception type; must surface to the LLM as an error string, not crash the turn
-                logger.warning("Room %s: Failed to send approval notification", room_id)
-                # A reply that claimed it meanwhile owns the answer; otherwise
-                # decline without a notice, so the missing-reply guard stays.
-                if registry.withdraw(entry):
-                    self._clear_pending_approval(room_id, entry)
-                    return PermissionResultDeny(
-                        message="Could not deliver approval prompt, tool use declined"
+        await set_permission_wait(tools, True)
+        try:
+            if tools:
+                try:
+                    await tools.send_notice(
+                        APPROVAL_REQUESTED_TEMPLATE.format(
+                            summary=summary, token=token
+                        ),
+                        mentions=mention,
                     )
+                except Exception:  # noqa: BLE001 -- tool calls may raise any exception type; must surface to the LLM as an error string, not crash the turn
+                    logger.warning(
+                        "Room %s: Failed to send approval notification", room_id
+                    )
+                    # A reply that claimed it meanwhile owns the answer; otherwise
+                    # decline without a notice, so the missing-reply guard stays.
+                    if registry.withdraw(entry):
+                        return PermissionResultDeny(
+                            message="Could not deliver approval prompt, tool use declined"
+                        )
+
+            # The request has been posted (or there was nowhere to post it) --
+            # either way, on_message must return now so Band's room loop can
+            # dispatch the reply that will resolve this wait. The turn is then
+            # judged at its real end (see _on_turn_complete).
+            if tools:
+                tools.turn.detach()
+            self._release_turn(room_id)
 
-        # The request has been posted (or there was nowhere to post it) --
-        # either way, on_message must return now so Band's room loop can
-        # dispatch the reply that will resolve this wait. The turn is then
-        # judged at its real end (see _on_turn_complete).
-        if tools:
-            tools.turn.detach()
-        self._release_turn(room_id)
-
-        try:
             decision = await registry.wait(
                 entry, pending.future, timeout_s=self._approvals.wait_timeout_s
             )
         finally:
             self._clear_pending_approval(room_id, entry)
+            await set_permission_wait(tools, self._open_approval_count(room_id) > 0)
 
         match decision:
             case Timeout.TIMED_OUT:
@@ -2062,8 +2082,7 @@ async def _handle_status_command(
         session_count = manager.get_session_count() if manager else 0
         session = manager.session(room_id) if manager else None
         session_id = (session.session_id if session else None) or "—"
-        registry = self._pending_approvals.get(room_id)
-        pending_count = registry.unclaimed_count() if registry else 0
+        pending_count = self._open_approval_count(room_id)
 
         lines = [
             "**Claude SDK Status**",
@@ -2120,6 +2139,10 @@ def _next_approval_token(self, room_id: str) -> str:
     def _new_approval_registry(self) -> DecisionRegistry[PendingApproval]:
         return DecisionRegistry(max_pending=self._approvals.max_pending_per_room)
 
+    def _open_approval_count(self, room_id: str) -> int:
+        registry = self._pending_approvals.get(room_id)
+        return registry.unclaimed_count() if registry else 0
+
     def _clear_pending_approval(
         self, room_id: str, entry: DecisionEntry[PendingApproval]
     ) -> None:

src/band/adapters/codex.py

@@ -101,6 +101,7 @@
     build_agent_failure,
     parse_plan_steps,
 )
+from band.runtime.activity import set_permission_wait
 from band.runtime.custom_tools import (
     CustomToolDef,
     custom_tool_to_openai_schema,
@@ -116,6 +117,12 @@
 )
 from band.runtime.formatters import strip_leading_mentions
 from band.runtime.prompts import COMMUNICATION_INSTRUCTIONS, render_system_prompt
+from band.runtime.toolactivity import (
+    end_tool_activity,
+    report_memory_activity,
+    report_tool_activity,
+    tool_activity,
+)
 from band.runtime.tools import (
     get_band_tool_category,
     image_block_placeholder,
@@ -292,6 +299,9 @@ class CodexSandboxMode(StrEnum):
     {CodexItemType.SUB_AGENT_ACTIVITY}
 )
 
+# Items that show as the turn's activity from item/started to item/completed.
+_ACTIVITY_ITEM_TYPES = _TOOL_ITEM_TYPES | {CodexItemType.CONTEXT_COMPACTION}
+
 _UNRENDERED_ITEM_TYPES = (
     _REQUESTED_TOOL_ITEM_TYPES | _MESSAGE_ITEM_TYPES | _ROUTING_ITEM_TYPES
 )
@@ -1697,10 +1707,10 @@ async def _process_turn_events(
 
                 if event.method == "item/started":
                     item = params.get("item")
-                    if (
-                        isinstance(item, dict)
-                        and item.get("type") == CodexItemType.AGENT_MESSAGE
-                    ):
+                    if not isinstance(item, dict):
+                        continue
+                    await self._show_item_started(tools, item)
+                    if item.get("type") == CodexItemType.AGENT_MESSAGE:
                         result.native_messages[str(item["id"])] = NativeMessage(
                             phase=item.get("phase"),
                         )
@@ -1739,6 +1749,8 @@ async def _process_turn_events(
                 if event.method == "item/completed":
                     item = params.get("item")
                     if isinstance(item, dict):
+                        if item.get("type") in _ACTIVITY_ITEM_TYPES:
+                            await end_tool_activity(tools)
                         if item.get("type") == CodexItemType.AGENT_MESSAGE:
                             message = result.native_messages.setdefault(
                                 str(item["id"]),
@@ -2548,9 +2560,10 @@ async def _handle_server_request(
             try:
                 custom_tool = find_custom_tool(self._custom_tools, tool_name)
                 if custom_tool:
-                    result = await execute_custom_tool(
-                        custom_tool, arguments, turn=tools.turn
-                    )
+                    async with tool_activity(tools, tool_name):
+                        result = await execute_custom_tool(
+                            custom_tool, arguments, turn=tools.turn
+                        )
                     success = True
                 else:
                     # Structured: a base tool (e.g. band_send_message) can fail without
@@ -2935,6 +2948,16 @@ async def _emit_item_completed_events(
                 exc_info=True,
             )
 
+    async def _show_item_started(
+        self, tools: AgentToolsProtocol, item: dict[str, Any]
+    ) -> None:
+        item_type = item.get("type", "")
+        if item_type == CodexItemType.CONTEXT_COMPACTION:
+            await report_memory_activity(tools)
+        elif item_type in _TOOL_ITEM_TYPES:
+            name = self._extract_tool_item(item_type, item).name
+            await report_tool_activity(tools, name)
+
     async def _emit_item_event(
         self,
         *,
@@ -3220,6 +3243,7 @@ async def _resolve_manual_approval(
                 self.config.max_pending_approvals_per_room,
             )
         try:
+            await set_permission_wait(tools, True)
             approval_msg = APPROVAL_REQUESTED_TEMPLATE.format(
                 summary=summary, token=token
             )
@@ -3303,6 +3327,7 @@ async def _resolve_manual_approval(
             return timeout_decision
         finally:
             self._clear_pending_approval(room_id, entry)
+            await set_permission_wait(tools, self._open_approval_count(room_id) > 0)
 
     async def _forward_raw_task_event(
         self,

src/band/integrations/mcp/engine.py

@@ -42,7 +42,6 @@
 from band.core.exceptions import BandToolError, MessagePostError, MessagePostKind
 from band.core.protocols import AgentToolsProtocol
 from band.core.tool_filter import sanitize_tool_schema
-from band.core.turn import Turn
 from band.core.types import Capability, EventMessageType, MessageType
 from band.runtime.custom_tools import (
     CustomToolDef,
@@ -572,12 +571,13 @@ async def execute(arguments: dict[str, Any]) -> Any:
 RoomToolResolver = Callable[[str], AgentToolsProtocol | None]
 
 
-def room_turn(get_tools: RoomToolResolver | None, chat_id: str | None) -> Turn | None:
-    """The turn of the room ``chat_id`` names, if the tool is room-bound."""
+def room_tools(
+    get_tools: RoomToolResolver | None, chat_id: str | None
+) -> AgentToolsProtocol | None:
+    """The tools of the room ``chat_id`` names, if the tool is room-bound."""
     if get_tools is None or chat_id is None:
         return None
-    tools = get_tools(chat_id)
-    return tools.turn if tools is not None else None
+    return get_tools(chat_id)
 
 
 def build_custom_tool_registration(
@@ -609,8 +609,11 @@ def build_custom_tool_registration(
     async def execute(arguments: dict[str, Any]) -> Any:
         kwargs = dict(arguments)
         chat_id = kwargs.pop(CHAT_ID_FIELD_NAME, None)
-        turn = room_turn(get_tools, chat_id)
-        result = await execute_custom_tool(tool_def, kwargs, turn=turn)
+        tools = room_tools(get_tools, chat_id)
+        async with tool_activity(tools, tool_name):
+            result = await execute_custom_tool(
+                tool_def, kwargs, turn=tools.turn if tools is not None else None
+            )
         return tool_result_text(result)
 
     return MCPToolRegistration(

src/band/testing/activity.py

@@ -41,5 +41,11 @@ async def update(self, activity: AgentActivityDetail) -> None:
     async def set_manual_wait(self, waiting: bool) -> None:
         self.events.append({"waiting": waiting})
 
+    @property
+    def waiting(self) -> bool:
+        """Whether the last reported wait says a person is being waited on."""
+        waits = [event["waiting"] for event in self.events if "waiting" in event]
+        return bool(waits) and waits[-1]
+
     def assert_completed_tool(self, name: str) -> None:
         assert self.events == [shown_tool(name), SHOWN_REASONING]

tests/adapters/claude_sdk/activity/__init__.py

@@ -0,0 +1 @@
+"""What a Claude SDK room's turn reports as its activity."""

tests/adapters/claude_sdk/activity/conftest.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+import pytest
+
+from band.testing.activity import ActivityRecorder
+
+
+@pytest.fixture
+def recorder() -> ActivityRecorder:
+    return ActivityRecorder()

tests/adapters/claude_sdk/activity/kit.py

@@ -0,0 +1,23 @@
+"""A turn that replies over Band's MCP server, then runs a native tool."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from band.runtime.tools.types import BandTool
+from band.testing.activity import SHOWN_REASONING, shown_tool
+from tests.adapters.claude_sdk.helpers import ClaudeRoom
+from tests.baseline.decisions import ModelDecision
+
+# The reply's MCP call ends before the CLI streams the Bash call, so the order
+# is fixed.
+REPLY_THEN_BASH_ACTIVITY: list[dict[str, Any]] = [
+    shown_tool(BandTool.SEND_MESSAGE),
+    SHOWN_REASONING,
+    shown_tool("Bash"),
+    SHOWN_REASONING,
+]
+
+
+def reply_then_bash(room: ClaudeRoom) -> list[ModelDecision]:
+    return [room.model_reply("Listing."), ModelDecision.call("Bash", command="ls")]

tests/adapters/claude_sdk/activity/test_narration_off.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from band.core.types import MessageType
+from band.testing.activity import ActivityRecorder
+from tests.adapters.claude_sdk.activity.kit import (
+    REPLY_THEN_BASH_ACTIVITY,
+    reply_then_bash,
+)
+from tests.adapters.claude_sdk.helpers import OpenRoom
+
+NARRATION = {MessageType.TOOL_CALL, MessageType.TOOL_RESULT}
+
+
+class TestActivityWithoutNarration:
+    async def test_a_native_tool_reports_activity_while_narration_is_off(
+        self, claude_room: OpenRoom, recorder: ActivityRecorder
+    ) -> None:
+        room = await claude_room(emit=())
+        recorder.observing(room.tools)
+        room.claude.script(reply_then_bash(room))
+
+        await room.send("List the files")
+
+        assert recorder.events == REPLY_THEN_BASH_ACTIVITY
+        assert NARRATION.isdisjoint(room.events)

tests/adapters/claude_sdk/activity/test_native_tools.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from band.testing.activity import ActivityRecorder
+from tests.adapters.claude_sdk.activity.kit import (
+    REPLY_THEN_BASH_ACTIVITY,
+    reply_then_bash,
+)
+from tests.adapters.claude_sdk.helpers import OpenRoom
+
+
+class TestClaudeNativeToolActivity:
+    async def test_a_native_tool_shows_until_its_result_and_band_tools_show_once(
+        self, claude_room: OpenRoom, recorder: ActivityRecorder
+    ) -> None:
+        """The reply's MCP call reports itself, so its streamed tool_use does
+        not report it again; a native tool reports from the stream."""
+        room = await claude_room()
+        recorder.observing(room.tools)
+        room.claude.script(reply_then_bash(room))
+
+        await room.send("List the files")
+
+        assert recorder.events == REPLY_THEN_BASH_ACTIVITY