feat(omp): report native tool and permission activity

The native adapter reports each host tool's start and end through the
shared activity helpers while tool-call narration is enabled, and marks
a permission wait around the resolver whatever the narration choice, so
the room's working indicator reflects native work and dialogs.

Author: bandzalkin

src/band/adapters/omp.py

@@ -74,6 +74,7 @@
     OmpUIRequest,
     OmpUIResponse,
 )
+from band.runtime.activity import set_permission_wait
 from band.runtime.custom_tools import (
     CustomToolDef,
     custom_tools_to_schemas,
@@ -86,6 +87,7 @@
     TURN_COMPLETION_GUIDANCE,
     render_system_prompt,
 )
+from band.runtime.toolactivity import end_tool_activity, report_tool_activity
 from band.runtime.tools import (
     BAND_MCP_SERVER_NAME,
     get_band_tool_category,
@@ -869,14 +871,18 @@ async def _permission(
         options = tuple(request.options or ())
         choice = None
         if self._resolve_permission is not None:
-            choice = await self._resolve_permission(
-                OmpPermissionRequest(
-                    room_id=room_id,
-                    session_id=session_id,
-                    title=request.title or "",
-                    options=options,
+            await set_permission_wait(tools, True)
+            try:
+                choice = await self._resolve_permission(
+                    OmpPermissionRequest(
+                        room_id=room_id,
+                        session_id=session_id,
+                        title=request.title or "",
+                        options=options,
+                    )
                 )
-            )
+            finally:
+                await set_permission_wait(tools, False)
         elif options == ("Approve", "Deny"):
             choice = "Approve"
         response = (
@@ -950,6 +956,7 @@ async def _emit_event(self, event: OmpEvent, turn: OmpTurn) -> None:
                 )
                 turn.calls[event.toolCallId] = (name, arguments)
                 if Emit.TOOL_CALLS in self.features.emit:
+                    await report_tool_activity(turn.tools, name)
                     await send_event_safe(
                         turn.tools,
                         self._encode_tool_call(name, arguments, event.toolCallId),
@@ -960,6 +967,8 @@ async def _emit_event(self, event: OmpEvent, turn: OmpTurn) -> None:
                 if call is None:
                     return
                 name, _arguments = call
+                if Emit.TOOL_CALLS in self.features.emit:
+                    await end_tool_activity(turn.tools)
                 if not self.config.inject_band_tools and not event.isError:
                     turn.tools.turn.record(turn_effect(name))
                 if Emit.TOOL_CALLS in self.features.emit:

tests/adapters/omp/kit.py

@@ -14,6 +14,7 @@
 
 from band.adapters.omp import OmpAdapter, OmpAdapterConfig
 from band.converters.omp import OmpSessionState
+from band.core.types import Emit
 from band.integrations.ompprocess import OmpMcpServer
 from band.integrations.omprpc import OmpRpcClient
 from band.integrations.ompwire import (
@@ -544,3 +545,38 @@ async def ask_turn(
 async def ask_frames(adapter: OmpAdapter) -> list[dict[str, JsonValue]]:
     checkpoint = await adapter._rooms["room-123"].runtime.request("checkpoint")
     return cast(list[dict[str, JsonValue]], checkpoint["frames"])
+
+
+EMIT_ON: tuple[Emit, ...] = (Emit.TOOL_CALLS,)
+EMIT_OFF: tuple[Emit, ...] = ()
+
+
+def make_adapter(
+    tmp_path: Path,
+    emit: tuple[Emit, ...],
+    resolve_permission=None,  # type: ignore[no-untyped-def]
+) -> OmpAdapter:
+    return OmpAdapter(
+        OmpAdapterConfig(cwd=str(tmp_path)),
+        emit=emit,
+        resolve_permission=resolve_permission,
+    )
+
+
+def tool_start(call_id: str, name: str, **args: object) -> OmpEvent:
+    return OmpEvent(
+        type="tool_execution_start",
+        toolCallId=call_id,
+        toolName=name,
+        args=args,
+    )
+
+
+def tool_end(call_id: str, name: str, *, is_error: bool = False) -> OmpEvent:
+    return OmpEvent(
+        type="tool_execution_end",
+        toolCallId=call_id,
+        toolName=name,
+        result=OmpToolResult(content=[OmpContent(type="text", text="private-result")]),
+        isError=is_error,
+    )

tests/adapters/omp/test_permission_activity.py

@@ -0,0 +1,98 @@
+"""Native OMP permission waits are reported as activity whatever the narration choice."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+import pytest
+
+from band.adapters.omp import OmpAdapter, OmpAdapterConfig, OmpPermissionRequest
+from band.core.types import Emit
+from band.integrations.ompwire import OmpUIRequest
+from band.testing.activity import ActivityRecorder
+from band.testing.fake_tools import FakeAgentTools
+from tests.adapters.omp.kit import EMIT_OFF, EMIT_ON, make_adapter
+
+
+def waiting(recorder: ActivityRecorder) -> bool:
+    """Whether the last manual-wait observation left the turn waiting."""
+    waits = [event["waiting"] for event in recorder.events if "waiting" in event]
+    return bool(waits) and waits[-1]
+
+
+class TestOmpPermissionActivity:
+    @pytest.mark.parametrize(
+        "emit", [EMIT_ON, EMIT_OFF], ids=["narration-on", "narration-off"]
+    )
+    async def test_wait_is_reported_whatever_the_narration_choice(
+        self, tmp_path: Path, emit: tuple[Emit, ...]
+    ) -> None:
+        recorder = ActivityRecorder()
+        tools = recorder.observing(FakeAgentTools())
+        release = asyncio.Event()
+        waiting_during: list[bool] = []
+
+        async def resolve(request: OmpPermissionRequest) -> str | None:
+            waiting_during.append(waiting(recorder))
+            await release.wait()
+            return "Approve" if request.options == ("Approve", "Deny") else None
+
+        adapter = OmpAdapter(
+            OmpAdapterConfig(cwd=str(tmp_path)),
+            emit=emit,
+            resolve_permission=resolve,
+        )
+        request = OmpUIRequest(
+            id="ui-1",
+            method="select",
+            title='Allow tool: Bash\nArguments: {"command": "private-command"}',
+            options=["Approve", "Deny"],
+        )
+        running = asyncio.create_task(
+            adapter._permission(request, "room", "sess", tools)
+        )
+        while not waiting_during:
+            await asyncio.sleep(0)
+        assert waiting(recorder)
+        release.set()
+        response = await running
+
+        assert response.value == "Approve"
+        assert waiting_during == [True]
+        assert not waiting(recorder)
+        assert recorder.events == [{"waiting": True}, {"waiting": False}]
+        assert not tools.events_sent
+
+    async def test_auto_approved_selection_reports_no_wait(
+        self, tmp_path: Path
+    ) -> None:
+        adapter = make_adapter(tmp_path, EMIT_OFF)
+        recorder = ActivityRecorder()
+        tools = recorder.observing(FakeAgentTools())
+        request = OmpUIRequest(
+            id="ui-1",
+            method="select",
+            title="Allow tool: Bash",
+            options=["Approve", "Deny"],
+        )
+
+        response = await adapter._permission(request, "room", "sess", tools)
+
+        assert response.value == "Approve"
+        assert recorder.events == []
+
+    async def test_non_select_ui_requests_report_no_wait(self, tmp_path: Path) -> None:
+        adapter = make_adapter(tmp_path, EMIT_OFF)
+        recorder = ActivityRecorder()
+        tools = recorder.observing(FakeAgentTools())
+
+        response = await adapter._permission(
+            OmpUIRequest(id="ui-1", method="confirm", title="private-question"),
+            "room",
+            "sess",
+            tools,
+        )
+
+        assert response.cancelled
+        assert recorder.events == []

tests/adapters/omp/test_tool_activity.py

@@ -0,0 +1,78 @@
+"""Native OMP tool activity is reported independently of narration."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from band.adapters.omp import OmpTurn
+from band.core.types import Emit
+from band.testing.activity import ActivityRecorder
+from band.testing.fake_tools import FakeAgentTools
+from tests.adapters.omp.kit import EMIT_OFF, EMIT_ON, make_adapter, tool_end, tool_start
+
+
+class TestOmpToolActivity:
+    @pytest.mark.parametrize(
+        "emit", [EMIT_ON, EMIT_OFF], ids=["narration-on", "narration-off"]
+    )
+    async def test_tool_activity_follows_the_narration_switch(
+        self, tmp_path: Path, emit: tuple[Emit, ...]
+    ) -> None:
+        adapter = make_adapter(tmp_path, emit)
+        recorder = ActivityRecorder()
+        tools = recorder.observing(FakeAgentTools())
+        turn = OmpTurn(tools=tools)
+
+        await adapter._emit_event(
+            tool_start("call-1", "Bash", command="private-command"), turn
+        )
+        await adapter._emit_event(tool_end("call-1", "Bash"), turn)
+
+        if emit:
+            recorder.assert_completed_tool("Bash")
+        else:
+            assert recorder.events == []
+        assert bool(tools.events_sent) is bool(emit)
+
+    async def test_only_the_name_reaches_the_recorder(self, tmp_path: Path) -> None:
+        adapter = make_adapter(tmp_path, EMIT_ON)
+        recorder = ActivityRecorder()
+        tools = recorder.observing(FakeAgentTools())
+        turn = OmpTurn(tools=tools)
+
+        await adapter._emit_event(
+            tool_start(
+                "call-1", "Write", file_path="private-path", content="private-content"
+            ),
+            turn,
+        )
+        assert recorder.events == [{"kind": "tool_call", "name": "Write"}]
+
+    async def test_latest_open_tool_wins(self, tmp_path: Path) -> None:
+        adapter = make_adapter(tmp_path, EMIT_ON)
+        recorder = ActivityRecorder()
+        tools = recorder.observing(FakeAgentTools())
+        turn = OmpTurn(tools=tools)
+
+        await adapter._emit_event(tool_start("call-a", "Read"), turn)
+        await adapter._emit_event(tool_start("call-b", "Bash"), turn)
+        assert recorder.events == [
+            {"kind": "tool_call", "name": "Read"},
+            {"kind": "tool_call", "name": "Bash"},
+        ]
+
+    async def test_end_of_untracked_call_changes_nothing(self, tmp_path: Path) -> None:
+        adapter = make_adapter(tmp_path, EMIT_ON)
+        recorder = ActivityRecorder()
+        tools = recorder.observing(FakeAgentTools())
+        turn = OmpTurn(tools=tools)
+
+        await adapter._emit_event(tool_end("never-started", "Ghost"), turn)
+        assert recorder.events == []
+
+        await adapter._emit_event(tool_start("call-1", "Bash"), turn)
+        await adapter._emit_event(tool_end("call-1", "Bash"), turn)
+        await adapter._emit_event(tool_end("call-1", "Bash"), turn)
+        recorder.assert_completed_tool("Bash")