feat(integrations): add standalone ACP preflight

.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 \

README.md

@@ -211,6 +211,28 @@ adapter = GeminiAdapter(GeminiAdapterConfig(model="gemini-3.8-flash"))
 
 Use [examples/run_agent.py](examples/run_agent.py) when you want one command that can switch between LangGraph, Pydantic AI, Anthropic, Claude SDK, Parlant, CrewAI, Codex, A2A bridge, and A2A gateway. Use the per-framework directories under [examples/](examples/) when you want the adapter-specific setup.
 
+### Host-side Harness Probes
+
+`await adapter.preflight(timeout_s=60)` reports a harness handshake as a
+`PreflightResult` (`ok`, `reason`, `remedy`). A successful handshake does not prove
+that the account can run inference. Adapters without a standalone handshake
+implementation explicitly report that it is unsupported.
+
+`await adapter.probe_models(timeout_s=60)` discovers a canonical `ModelCatalog`
+without creating or selecting a Band room. Unsupported adapters raise
+`BandConfigError`; an empty catalog is not a substitute for unsupported discovery.
+The existing room-facing `list_models` hook remains separate.
+
+Catalog entries preserve advertised provider and default metadata. An entry's
+`efforts=None` means the harness did not report effort support; `efforts=()` means
+it explicitly offers none. `current_model` is populated only when the harness
+reports an actual selection.
+
+Timeouts bound how long a caller waits, not ownership of a partially started
+transport. Always call `await adapter.cleanup_all()` when disposing of an adapter.
+An incomplete close raises `CleanupIncompleteError` and retains its resources for
+a later cleanup attempt rather than reporting successful disposal.
+
 ### Process Signals
 
 `agent.run()` and `agent.run_forever()` stop the agent on SIGTERM/SIGINT and restore the process's previous handlers when they return. A host that owns its process signals (a launchd/systemd service, a desktop app, a test runner) passes `install_signal_handlers=False` and calls `agent.stop()` from its own handler; `run_forever` then returns. `GracefulShutdown` is a ready-made handler that does this, and `run_with_graceful_shutdown(agent)` wraps it. Put `GracefulShutdown` outside `agent` so its handlers are registered before `start()` and a mid-initialize SIGTERM can still tear down:

docs/acp.md

@@ -24,6 +24,44 @@ adapter = ACPClientAdapter(config)
 assert adapter.config.command == ("codex-acp",)
 ```
 
+## Host preflight
+
+`await adapter.preflight(timeout_s=60.0)` on `ACPClientAdapter`,
+`CursorACPAdapter` or `CopilotACPAdapter` starts a throwaway ACP process,
+initializes it, and closes it without a Band connection, room, room workspace,
+ACP session or model turn. It uses the configured executable, backend credential
+environment, client capabilities and protocol options. A configured `cwd` is
+the probe's working directory itself, not a root for a fabricated room.
+This ACP operation is separate from the native OMP adapter's protocol.
+
+When `auth_method` is set, preflight also calls ACP `authenticate`. This can open
+an interactive login flow or change stored authentication state; leave it unset
+when a host must not request login. Even a successful initialize/authenticate
+does not establish inference entitlement, MCP connectivity, session creation or
+configured model/effort readiness. Session catalogs and model selection are
+still checked when a real room opens a session.
+
+The result is a `PreflightResult`: `ok`, with a safe `reason` and `remedy` on an
+ordinary handshake failure. `timeout_s` bounds handshake reporting and, separately,
+cleanup reporting. Cancellation propagates. `CleanupIncompleteError` is not a
+failed handshake result: the adapter still owns a process or late startup.
+Call `await adapter.cleanup_all()` again to settle it before discarding the
+adapter. Probe stderr diagnostics report line lengths and process exit metadata,
+not raw child output or configured command arguments. Third-party protocol
+libraries can emit their own diagnostics.
+
+```python notest
+from band.adapters.copilot_acp import CopilotACPAdapter
+
+adapter = CopilotACPAdapter()
+try:
+    result = await adapter.preflight(timeout_s=15.0)
+    if not result.ok:
+        raise RuntimeError(f"{result.reason} {result.remedy}")
+finally:
+    await adapter.cleanup_all()
+```
+
 ## Band tools
 
 - **Injected tools are bound to the room.** With `inject_band_tools` on (the

docs/adapters/codex.md

@@ -38,9 +38,11 @@ For existing integrations, follow the [SDK 4.x to 5.x migration](../acp.md#upgra
   On Windows, the client owns the helper and its process-tree job from startup
   through cleanup. It starts the interpreter image directly so a virtual-environment
   launcher cannot hide another helper process. Cleanup confirms the helper has
-  exited before terminating and draining the job; an empty job alone does not
-  establish helper exit.
-  Failed cleanup retains that same owner and job for retry.
+  exited, seals the job against new children, and retains member process handles
+  before requesting termination. It waits for those handles to signal physical
+  exit as well as empty job accounting; accounting alone does not prove directory
+  and file handles were released. Failed cleanup retains that same owner, job,
+  and any pending process handles for retry.
 - **Leaving is final.** Leaving a room stops it admitting work at once. A
   leave that fails or runs over keeps the process, thread and workspace claim,
   and the room refuses work until a retried leave succeeds; a later rejoin

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