feat: deliver compatible serialized activity detail

docs/turn-outcome.md

@@ -426,3 +426,20 @@ 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. Each turn keeps
+its own detail, starting at reasoning; the room shows the owner's detail. An
+aborted owner's stale detail is never sent: its reports while it unwinds carry
+no detail, which the platform shows as reasoning. Reporting what a turn is doing
+never starts reporting, changes ownership, or waits on the network: a visible
+change wakes the reporter's keep-alive, which sends the latest detail, so
+changes made while a report is in flight coalesce. 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/platform/activity.py

@@ -0,0 +1,67 @@
+"""Process-wide negotiation for optional room activity detail.
+
+Every agent API response names its version in ``X-API-Version``. Detail is sent
+once any activity response reported a supporting version. A detail rejection on
+any link permanently downgrades every link; boolean liveness remains available.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Mapping
+from threading import Lock
+
+logger = logging.getLogger(__name__)
+# The first agent API that accepts ``activity``; older ones reject it with 422.
+MINIMUM_ACTIVITY_VERSION = (1, 25, 0)
+API_VERSION_HEADER = "x-api-version"
+RELEASE_VERSION = re.compile(r"([0-9]+)\.([0-9]+)\.([0-9]+)")
+
+
+def _supports_detail(version: str | None) -> bool:
+    if version is None:
+        return False
+    match = RELEASE_VERSION.fullmatch(version)
+    return (
+        match is not None
+        and tuple(map(int, match.groups())) >= MINIMUM_ACTIVITY_VERSION
+    )
+
+
+class ActivitySupport:
+    """Learn detail support from response headers; a rejection latches it off."""
+
+    def __init__(self) -> None:
+        self._lock = Lock()
+        self._learned = False
+        self._disabled = False
+
+    @property
+    def supported(self) -> bool:
+        with self._lock:
+            return self._learned and not self._disabled
+
+    def observe(self, headers: Mapping[str, str]) -> None:
+        version = headers.get(API_VERSION_HEADER)
+        if not _supports_detail(version):
+            return
+        with self._lock:
+            newly_learned = not self._learned
+            self._learned = True
+        if newly_learned:
+            logger.info("Agent API %s supports activity detail", version)
+
+    def disable(self, *, room_id: str) -> None:
+        with self._lock:
+            already_disabled = self._disabled
+            self._disabled = True
+        if not already_disabled:
+            logger.info(
+                "Activity report with detail returned 422 in room %s; "
+                "using boolean reporting for this process",
+                room_id,
+            )
+
+
+activity_support = ActivitySupport()

src/band/platform/link.py

@@ -13,6 +13,7 @@
 from functools import partial
 from typing import TYPE_CHECKING
 
+from band_rest.types.agent_activity_detail import AgentActivityDetail
 from band_sdk_core import AgentTopicKind, SessionState
 
 from band.client.rest import AsyncRestClient
@@ -647,13 +648,15 @@ async def report_activity(
         room_id: str,
         working: bool,
         *,
+        activity: AgentActivityDetail | None = None,
         timeout_seconds: int = 2,
         on_stopped: Callable[[], None] | None = None,
     ) -> bool:
         return await self._messages.report_activity(
             self.rest,
             room_id,
             working,
+            activity=activity,
             timeout_seconds=timeout_seconds,
             on_stopped=on_stopped,
         )

src/band/platform/message_lifecycle.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 from collections import OrderedDict
 from collections.abc import Awaitable, Callable
@@ -14,9 +15,14 @@
 from http import HTTPStatus
 from typing import Any, TypeVar
 
+from band_rest.agent_api_activity.types.report_agent_chat_activity_response import (
+    ReportAgentChatActivityResponse,
+)
 from band_rest.core.api_error import ApiError
 from band_rest.core.http_response import AsyncHttpResponse
 from band_rest.core.request_options import RequestOptions
+from band_rest.types.agent_activity_detail import AgentActivityDetail
+from band_rest.types.agent_activity_response import AgentActivityResponse
 from band_rest.types.chat_message import ChatMessage
 from band_rest.types.error import Error as PlatformError
 from band_rest.types.validation_error import ValidationError as ErrorEnvelope
@@ -32,6 +38,7 @@
 from band.core.exceptions import RoomExecutionStoppedError
 from band.core.settlement import SettlementResult
 from band.core.types import metadata_to_dict
+from band.platform.activity import activity_support
 from band.runtime.messages import PlatformMessage
 
 logger = logging.getLogger(__name__)
@@ -109,6 +116,46 @@ def _failure_result(error: Exception) -> SettlementResult:
             return SettlementResult.UNCONFIRMED
 
 
+async def _post_activity(
+    rest: AsyncRestClient,
+    room_id: str,
+    working: bool,
+    activity: AgentActivityDetail | None,
+    timeout_seconds: int,
+) -> AgentActivityResponse:
+    detail = activity if activity_support.supported else None
+    try:
+        response = await _send_activity(rest, room_id, working, detail, timeout_seconds)
+    except ApiError as error:
+        if detail is None or error.status_code != HTTPStatus.UNPROCESSABLE_ENTITY:
+            raise
+        activity_support.disable(room_id=room_id)
+        response = await _send_activity(rest, room_id, working, None, timeout_seconds)
+    activity_support.observe(response.headers)
+    return response.data.data
+
+
+async def _send_activity(
+    rest: AsyncRestClient,
+    room_id: str,
+    working: bool,
+    activity: AgentActivityDetail | None,
+    timeout_seconds: int,
+) -> AsyncHttpResponse[ReportAgentChatActivityResponse]:
+    kwargs: dict[str, Any] = {}
+    if activity is not None:
+        kwargs["activity"] = activity
+    return await rest.agent_api_activity.with_raw_response.report_agent_chat_activity(
+        chat_id=room_id,
+        working=working,
+        request_options={
+            **NO_RETRY_REQUEST_OPTIONS,
+            "timeout_in_seconds": timeout_seconds,
+        },
+        **kwargs,
+    )
+
+
 def _platform_message(item: ChatMessage, room_id: str) -> PlatformMessage:
     return PlatformMessage(
         id=item.id,
@@ -306,38 +353,38 @@ async def report_activity(
         room_id: str,
         working: bool,
         *,
+        activity: AgentActivityDetail | None = None,
         timeout_seconds: int = 2,
         on_stopped: Callable[[], None] | None = None,
     ) -> bool:
         """
-        Report the agent's boolean working state for a room's execution.
+        Report the agent's working state and activity for a room's execution.
 
         ``working=True`` while a reasoning cycle is active (refreshed on a
         keep-alive cadence), ``False`` once it ends. Request failures are
         swallowed and returned as ``False``, since the platform's TTL is
         the backstop and activity reporting must never break message
         processing.
 
-        ``timeout_seconds`` bounds each POST so a slow/half-open endpoint
-        can't stall the keep-alive or wedge teardown. Retries are off: a
-        dropped keep-alive gets re-sent next cadence tick, and a dropped
-        ``false`` is cleared by the platform TTL either way — retrying would
-        only add latency.
+        Detail is attached once a response reported a supporting agent API
+        version. One deadline bounds the POST and its sole compatibility retry:
+        a rejected detail permanently downgrades the process to boolean
+        reports; other failures wait for the next keep-alive or platform TTL.
 
         A successful ``working=True`` report echoed as ``False`` identifies a
         stopped execution. Notify ``on_stopped`` separately from the request's
         success boolean; transport failures never imply STOP.
         """
         try:
-            response = await rest.agent_api_activity.report_agent_chat_activity(
-                chat_id=room_id,
-                working=working,
-                request_options={
-                    **NO_RETRY_REQUEST_OPTIONS,
-                    "timeout_in_seconds": timeout_seconds,
-                },
-            )
-            stopped = working and response.data.working is False
+            async with asyncio.timeout(timeout_seconds):
+                response = await _post_activity(
+                    rest,
+                    room_id,
+                    working,
+                    activity if working else None,
+                    timeout_seconds,
+                )
+            stopped = working and response.working is False
         except Exception as e:  # noqa: BLE001 -- best-effort event emission must not crash the turn/link
             if not self._activity_report_failing:
                 self._activity_report_failing = True

src/band/runtime/activity.py

@@ -0,0 +1,80 @@
+"""Names-only room activity detail, the per-turn state that holds it, and the
+producers a turn reports it through."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Protocol
+
+from band_rest.types.agent_activity_detail import AgentActivityDetail
+
+UNKNOWN_ACTIVITY_TOOL_NAME = "unknown"
+TOOL_CALL = "tool_call"
+REASONING = AgentActivityDetail(kind="reasoning")
+MEMORY = AgentActivityDetail(kind="memory")
+PERMISSION = AgentActivityDetail(kind="permission")
+
+
+class ActivityObserver(Protocol):
+    async def update(self, activity: AgentActivityDetail) -> None: ...
+
+    async def set_manual_wait(self, waiting: bool) -> None: ...
+
+
+def tool_call_activity(name: str | None) -> AgentActivityDetail:
+    """The platform requires a name on every ``tool_call`` detail."""
+    return AgentActivityDetail(kind=TOOL_CALL, name=name or UNKNOWN_ACTIVITY_TOOL_NAME)
+
+
+def names_only(activity: AgentActivityDetail) -> AgentActivityDetail:
+    """Keep only what the room may see: the kind and, for a tool, its name."""
+    if activity.kind == TOOL_CALL:
+        return tool_call_activity(activity.name)
+    return AgentActivityDetail(kind=activity.kind)
+
+
+@dataclass
+class TurnActivity:
+    """What one turn is doing; a person's wait overrides its progress."""
+
+    progress: AgentActivityDetail = REASONING
+    waiting: bool = False
+
+    @property
+    def detail(self) -> AgentActivityDetail:
+        return PERMISSION if self.waiting else self.progress
+
+    def apply(
+        self,
+        *,
+        activity: AgentActivityDetail | None = None,
+        waiting: bool | None = None,
+    ) -> None:
+        if activity is not None:
+            self.progress = names_only(activity)
+        if waiting is not None:
+            self.waiting = waiting
+
+
+def _observer(tools: object) -> ActivityObserver | None:
+    return getattr(tools, "_activity_observer", None)
+
+
+async def report_tool_activity(tools: object, name: str) -> None:
+    if observer := _observer(tools):
+        await observer.update(tool_call_activity(name))
+
+
+async def end_tool_activity(tools: object) -> None:
+    if observer := _observer(tools):
+        await observer.update(REASONING)
+
+
+async def report_memory_activity(tools: object) -> None:
+    if observer := _observer(tools):
+        await observer.update(MEMORY)
+
+
+async def set_permission_wait(tools: object, waiting: bool) -> None:
+    if observer := _observer(tools):
+        await observer.set_manual_wait(waiting)

src/band/runtime/cycle.py

@@ -7,6 +7,7 @@
 
 from band.client.streaming import ControlMode
 from band.core.turn_lifecycle import TurnTask, terminal_deadline
+from band.runtime.activity import TurnActivity
 
 
 @dataclass(eq=False)
@@ -28,6 +29,7 @@ class TurnScope:
     shutdown: bool = False
     terminated: asyncio.Event = field(default_factory=asyncio.Event)
     workers: set[TurnTask] = field(default_factory=set)
+    activity: TurnActivity = field(default_factory=TurnActivity)
 
     @property
     def stop_observed(self) -> bool: