feat: gate activity tool names on the tool-call emit choice

.github/workflows/ci.yml

@@ -163,6 +163,11 @@ jobs:
         BAND_ALLOW_MISSING_FRAMEWORKS: "1"
       run: uv run --all-packages pytest tests/e2e/baseline/ --collect-only -q --no-cov
 
+    - name: Run CrewAI activity tests
+      env:
+        BAND_ALLOW_MISSING_FRAMEWORKS: "1"
+      run: uv run --all-packages pytest tests/integrations/crewai/activity/
+
     - name: Run crewai tests
       env:
         # dev-crewai venv lacks langgraph/anthropic/parlant/pydantic-ai/etc.;
@@ -171,7 +176,7 @@ jobs:
       # Only the tests that *need* this venv. Every other crewai test fakes the
       # package through sys.modules, so the `test` job above already runs it and
       # a second pass here would add no signal — just a path to forget.
-      # tests/framework_conformance/test_crewai_job_coverage.py keeps this honest.
+      # tests/framework_conformance/crewaicoverage/ keeps this honest.
       run: |
         uv run --all-packages pytest \
           tests/adapters/test_crewai_flow_phase3.py \
@@ -233,6 +238,12 @@ jobs:
         BAND_TEST_EXTRA: "dev-strands-nooa"
       run: uv run --all-packages pytest tests/e2e/baseline/ --collect-only -q --no-cov
 
+    - name: Run Strands activity tests
+      env:
+        BAND_ALLOW_MISSING_FRAMEWORKS: "1"
+        BAND_TEST_EXTRA: "dev-strands-nooa"
+      run: uv run --all-packages pytest tests/adapters/strands/activity/
+
     - name: Run Strands and NOOA tests
       env:
         # dev-strands-nooa lacks modern framework packages;
@@ -302,6 +313,11 @@ jobs:
         BAND_ALLOW_MISSING_FRAMEWORKS: "1"
       run: uv run --all-packages pytest tests/e2e/baseline/ --collect-only -q --no-cov
 
+    - name: Run Parlant activity tests
+      env:
+        BAND_ALLOW_MISSING_FRAMEWORKS: "1"
+      run: uv run --all-packages pytest tests/integrations/parlant/activity/
+
     - name: Run parlant tests
       env:
         # dev-parlant venv lacks langgraph/anthropic-adapter/pydantic-ai/etc.;
@@ -310,7 +326,7 @@ jobs:
       # Only the tests that *need* this venv. Every other parlant test fakes the
       # package through sys.modules, so the `test` job above already runs it and
       # a second pass here would add no signal — just a path to forget.
-      # tests/framework_conformance/test_parlant_job_coverage.py keeps this honest.
+      # tests/framework_conformance/parlantcoverage/ keeps this honest.
       run: |
         uv run --all-packages pytest \
           tests/integrations/parlant/test_catalog.py \

docs/turn-outcome.md

@@ -424,3 +424,15 @@ once. It does not rewrite the original delivery's settled acknowledgement.
 Explicit interruption invalidates its activity; leave and shutdown clear the
 indicator while adapter residency retains responsibility for released workers.
 The authoritative idle report finishes before a replacement reports working.
+
+`SessionConfig.enable_working_state` controls both the indicator and activity
+detail, independently of chat narration and tool capabilities. Changes and
+keep-alives share one serialized reporter. The duration cap stops both kinds
+of refresh without cancelling provider work; actual completion still clears.
+
+Detail compatibility is learned from the `X-API-Version` header the platform
+puts on every activity response. Reports stay boolean until a response in the
+process advertises agent API 1.25.0 or later; a missing or older version keeps
+boolean reporting. A detail-bearing HTTP 422 permanently disables detail across
+all links in the process and retries that report once without detail, within
+its original deadline.

src/band/adapters/agno.py

@@ -37,6 +37,7 @@
     is_image_passthrough_result,
     redact_tool_call_args,
 )
+from band.runtime.working_state import end_tool_activity, report_tool_activity
 
 try:
     from agno.models.message import Message
@@ -451,10 +452,7 @@ async def _run_agent(
     ) -> RunOutput | None:
         """Run the Agno agent with the room's tools bound for this call.
 
-        When ``Emit.TOOL_CALLS`` is enabled the run is streamed so tool_call /
-        tool_result events are emitted *as each tool runs* (see
-        :meth:`_run_streamed`), matching the other adapters' live reporting.
-        Otherwise it runs non-streaming, exactly as before.
+        Consume provider progress as it happens, independently of chat narration.
         """
         try:
             agent = self._agent
@@ -469,17 +467,14 @@ async def _run_agent(
                 session_id,
             )
             with _bind_room_tools(tools):
-                if Emit.TOOL_CALLS in self.features.emit:
-                    response = await self._run_streamed(
-                        agent,
-                        messages,
-                        tools,
-                        session_id=session_id,
-                        room_id=room_id,
-                        msg_id=msg_id,
-                    )
-                else:
-                    response = await agent.arun(input=messages, session_id=session_id)
+                response = await self._run_streamed(
+                    agent,
+                    messages,
+                    tools,
+                    session_id=session_id,
+                    room_id=room_id,
+                    msg_id=msg_id,
+                )
             # Agno swallows run exceptions into a normal-looking return value
             # (status=error); surface it so the shared except path below treats
             # the turn as failed rather than as a silent empty reply.
@@ -532,9 +527,14 @@ async def _run_streamed(
                 # None, which would read as a successful turn with no output.
                 raise AgnoRunError(_error_summary(item.content))
             else:
-                await self._emit_stream_event(
-                    item, tools, room_id=room_id, msg_id=msg_id
-                )
+                if isinstance(item, ToolCallStartedEvent) and item.tool is not None:
+                    await report_tool_activity(tools, item.tool.tool_name or "")
+                elif isinstance(item, ToolCallCompletedEvent):
+                    await end_tool_activity(tools)
+                if Emit.TOOL_CALLS in self.features.emit:
+                    await self._emit_stream_event(
+                        item, tools, room_id=room_id, msg_id=msg_id
+                    )
         return final
 
     def _persist_turn(self, room_id: str, response: RunOutput) -> None:

src/band/adapters/anthropic.py

@@ -40,6 +40,7 @@
     redact_tool_call_args,
 )
 from band.runtime.tools.schema import is_failed_tool_output
+from band.runtime.working_state import tool_activity
 
 logger = logging.getLogger(__name__)
 
@@ -478,61 +479,62 @@ async def _process_tool_calls(
 
             logger.debug("Executing tool: %s with input: %s", tool_name, tool_input)
 
-            # Report tool call if enabled (JSON format with tool_call_id for linking)
-            # Best-effort: event reporting must never crash tool execution
-            if Emit.TOOL_CALLS in self.features.emit:
+            async with tool_activity(tools, tool_name):
+                # Report tool call if enabled (JSON format with tool_call_id for linking)
+                # Best-effort: event reporting must never crash tool execution
+                if Emit.TOOL_CALLS in self.features.emit:
+                    try:
+                        await tools.send_event(
+                            content=json.dumps(
+                                {
+                                    ToolEventKey.NAME: tool_name,
+                                    ToolEventKey.ARGS: redact_tool_call_args(
+                                        tool_name, tool_input
+                                    ),
+                                    ToolEventKey.TOOL_CALL_ID: tool_use_id,
+                                }
+                            ),
+                            message_type="tool_call",
+                        )
+                    except Exception as e:  # noqa: BLE001 -- tool calls may raise any exception type; must surface to the LLM as an error string, not crash the turn
+                        logger.warning(
+                            "Failed to send tool_call event: %s",
+                            e,
+                        )
+
+                # Execute tool (check custom tools first, then platform tools)
                 try:
-                    await tools.send_event(
-                        content=json.dumps(
-                            {
-                                ToolEventKey.NAME: tool_name,
-                                ToolEventKey.ARGS: redact_tool_call_args(
-                                    tool_name, tool_input
-                                ),
-                                ToolEventKey.TOOL_CALL_ID: tool_use_id,
-                            }
-                        ),
-                        message_type="tool_call",
-                    )
+                    custom_tool = find_custom_tool(self._custom_tools, tool_name)
+                    if custom_tool:
+                        result = await execute_custom_tool(
+                            custom_tool, tool_input, turn=tools.turn
+                        )
+                        succeeded = not is_failed_tool_output(result)
+                    else:
+                        outcome = await tools.execute_tool_call_structured(
+                            tool_name, tool_input
+                        )
+                        result = outcome.value
+                        succeeded = outcome.ok
+                    if succeeded and tool_name == BandTool.NO_REPLY:
+                        declined = True
+                    if not succeeded:
+                        failed = True
+                    if is_image_passthrough_result(tool_name, result):
+                        content = _image_tool_result_content(result)
+                        result_str = image_block_placeholder(len(content))
+                    else:
+                        content = result_str = (
+                            json.dumps(result, default=str)
+                            if not isinstance(result, str)
+                            else result
+                        )
+                    is_error = False
                 except Exception as e:  # noqa: BLE001 -- tool calls may raise any exception type; must surface to the LLM as an error string, not crash the turn
-                    logger.warning(
-                        "Failed to send tool_call event: %s",
-                        e,
-                    )
-
-            # Execute tool (check custom tools first, then platform tools)
-            try:
-                custom_tool = find_custom_tool(self._custom_tools, tool_name)
-                if custom_tool:
-                    result = await execute_custom_tool(
-                        custom_tool, tool_input, turn=tools.turn
-                    )
-                    succeeded = not is_failed_tool_output(result)
-                else:
-                    outcome = await tools.execute_tool_call_structured(
-                        tool_name, tool_input
-                    )
-                    result = outcome.value
-                    succeeded = outcome.ok
-                if succeeded and tool_name == BandTool.NO_REPLY:
-                    declined = True
-                if not succeeded:
                     failed = True
-                if is_image_passthrough_result(tool_name, result):
-                    content = _image_tool_result_content(result)
-                    result_str = image_block_placeholder(len(content))
-                else:
-                    content = result_str = (
-                        json.dumps(result, default=str)
-                        if not isinstance(result, str)
-                        else result
-                    )
-                is_error = False
-            except Exception as e:  # noqa: BLE001 -- tool calls may raise any exception type; must surface to the LLM as an error string, not crash the turn
-                failed = True
-                content = result_str = f"Error: {e}"
-                is_error = True
-                logger.error("Tool %s failed: %s", tool_name, e)
+                    content = result_str = f"Error: {e}"
+                    is_error = True
+                    logger.error("Tool %s failed: %s", tool_name, e)
 
             # Report tool result if enabled (JSON format with tool_call_id for linking)
             # Best-effort: event reporting must never crash tool execution

src/band/adapters/claude_sdk.py

@@ -23,6 +23,12 @@
 from functools import partial
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, NoReturn, Self, cast
 
+from band.runtime.working_state import (
+    end_tool_activity,
+    report_tool_activity,
+    set_permission_wait,
+)
+
 try:
     from claude_agent_sdk import (  # type: ignore[import-not-found]
         AssistantMessage,
@@ -1406,6 +1412,7 @@ async def _on_tool_use(
         # namespaces our tools mcp__band__*; see _semantic_tool_name).
         tool_name = self._semantic_tool_name(block.name)
         pending_tool_names[block.id] = tool_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,6 +1562,7 @@ 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).
         """
+        await end_tool_activity(tools)
         result_tool_name = pending_tool_names.pop(block.tool_use_id, None)
         await self._narrate_tool_result(block, result_tool_name, room_id, tools)
 
@@ -1858,48 +1866,56 @@ 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"
+        try:
+            if tools:
+                await set_permission_wait(tools, True)
+                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
+                entry,
+                pending.future,
+                timeout_s=self._approvals.wait_timeout_s,
+                retain_entry=True,
             )
+            match decision:
+                case Timeout.TIMED_OUT:
+                    return await self._apply_approval_timeout(
+                        tools, room_id, token, mention, tool_use_id
+                    )
+                case ApprovalReply():
+                    return await self._apply_approval_reply(
+                        tools, room_id, token, decision, tool_use_id
+                    )
+            # Forced by eviction or room teardown: no notice names this call.
+            return PermissionResultDeny(message="User declined tool use")
         finally:
             self._clear_pending_approval(room_id, entry)
-
-        match decision:
-            case Timeout.TIMED_OUT:
-                return await self._apply_approval_timeout(
-                    tools, room_id, token, mention, tool_use_id
-                )
-            case ApprovalReply():
-                return await self._apply_approval_reply(
-                    tools, room_id, token, decision, tool_use_id
-                )
-        # Forced by eviction or room teardown: no notice names this call.
-        return PermissionResultDeny(message="User declined tool use")
+            if tools:
+                await set_permission_wait(tools, bool(registry))
 
     async def _apply_approval_reply(
         self,