diff --git a/docs/turn-outcome.md b/docs/turn-outcome.md
index 54f7541ed..6be76ef6c 100644
--- a/docs/turn-outcome.md
+++ b/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.
diff --git a/src/band/platform/activity.py b/src/band/platform/activity.py
new file mode 100644
index 000000000..41a57b0f5
--- /dev/null
+++ b/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()
diff --git a/src/band/platform/link.py b/src/band/platform/link.py
index c512952d9..f85634185 100644
--- a/src/band/platform/link.py
+++ b/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,6 +648,7 @@ async def report_activity(
         room_id: str,
         working: bool,
         *,
+        activity: AgentActivityDetail | None = None,
         timeout_seconds: int = 2,
         on_stopped: Callable[[], None] | None = None,
     ) -> bool:
@@ -654,6 +656,7 @@ async def report_activity(
             self.rest,
             room_id,
             working,
+            activity=activity,
             timeout_seconds=timeout_seconds,
             on_stopped=on_stopped,
         )
diff --git a/src/band/platform/message_lifecycle.py b/src/band/platform/message_lifecycle.py
index 2e48b52fb..a2b55bdeb 100644
--- a/src/band/platform/message_lifecycle.py
+++ b/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,11 +353,12 @@ 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
@@ -318,26 +366,25 @@ async def report_activity(
         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
diff --git a/src/band/runtime/activity.py b/src/band/runtime/activity.py
new file mode 100644
index 000000000..62de696b3
--- /dev/null
+++ b/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)
diff --git a/src/band/runtime/cycle.py b/src/band/runtime/cycle.py
index 789ac995f..207f89beb 100644
--- a/src/band/runtime/cycle.py
+++ b/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:
diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py
index 923d91c7a..499b65d6e 100644
--- a/src/band/runtime/execution.py
+++ b/src/band/runtime/execution.py
@@ -19,6 +19,7 @@
 import logging
 from collections import deque
 from collections.abc import Awaitable, Callable, Iterator
+from dataclasses import dataclass
 from datetime import UTC, datetime
 from enum import Enum, StrEnum
 from functools import partial
@@ -30,6 +31,7 @@
     runtime_checkable,
 )
 
+from band_rest.types.agent_activity_detail import AgentActivityDetail
 from band_sdk_core import (
     AgentFailure,
     ClaimRegistry,
@@ -74,6 +76,7 @@
     PlatformEvent,
     ReconnectedEvent,
 )
+from band.runtime.activity import ActivityObserver
 from band.runtime.callbacks import ParticipantAddedCallback, ParticipantRemovedCallback
 from band.runtime.context import ConversationContext
 from band.runtime.context_serialization import context_item_to_dict
@@ -282,6 +285,18 @@ async def resume_room(self) -> None:
 ExecutionHandler = Callable[["ExecutionContext", PlatformEvent], Awaitable[None]]
 
 
+@dataclass(frozen=True)
+class ScopeActivityObserver:
+    context: ExecutionContext
+    scope: TurnScope
+
+    async def update(self, activity: AgentActivityDetail) -> None:
+        await self.context.observe_activity(self.scope, activity=activity)
+
+    async def set_manual_wait(self, waiting: bool) -> None:
+        await self.context.observe_activity(self.scope, waiting=waiting)
+
+
 class ExecutionContext:
     """
     Default execution: context accumulation model.
@@ -350,7 +365,7 @@ def __init__(
         self._on_participant_removed = on_participant_removed
         self.hub_room_id = hub_room_id
 
-        # Per-room boolean working-state reporter. Disabled for the contact-hub
+        # Per-room working-state reporter. Disabled for the contact-hub
         # room: it's internal housekeeping, not a peer-facing conversation, so no
         # counterpart watches a "Reasoning…" indicator there.
         self._working_reporter = WorkingStateReporter(
@@ -432,11 +447,6 @@ def __init__(
         # The first stop's graceful result; a retry only settles retained work.
         self._stop_graceful = True
 
-        # Optional seam for clearing user-visible activity state ("reasoning…")
-        # when a cycle is interrupted/stopped. Filled by the activity-signal
-        # work; a no-op (None) here.
-        self._on_activity_clear: Callable[[], Awaitable[None]] | None = None
-
     @property
     def thread_id(self) -> str:
         """LangGraph thread_id = room_id."""
@@ -871,17 +881,6 @@ async def _stop_retained_tasks(self, deadline: float) -> None:
                 "task(s) still running"
             )
 
-    async def _clear_activity(self) -> None:
-        """Invoke the optional activity-clear seam after an aborted cycle."""
-        if self._on_activity_clear is None:
-            return
-        try:
-            await self._on_activity_clear()
-        except Exception:
-            logger.exception(
-                "ExecutionContext %s: activity-clear hook failed", self.room_id
-            )
-
     # --- Participant management ---
 
     def add_participant(self, participant: dict[str, Any]) -> bool:
@@ -1874,10 +1873,11 @@ async def _invoke_handler(self, event: PlatformEvent) -> None:
                 await self._on_execute(self, event)
 
     async def _report_working_state(self, working: bool) -> bool:
-        """Report the room's boolean working state (wired into the reporter)."""
+        """Report working state and visible activity (wired into the reporter)."""
         return await self.link.report_activity(
             self.room_id,
             working,
+            activity=self._visible_activity,
             timeout_seconds=self.config.working_request_timeout_seconds,
             on_stopped=partial(
                 self._observe_activity_stop,
@@ -1944,6 +1944,31 @@ async def _start_activity(self, scope: TurnScope) -> None:
         self._activity_owner = scope
         await self._working_reporter.start()
 
+    def activity_observer(self, scope: TurnScope | None) -> ActivityObserver | None:
+        return ScopeActivityObserver(self, scope) if scope is not None else None
+
+    @property
+    def _visible_activity(self) -> AgentActivityDetail | None:
+        """The owner's detail, or none for an aborted owner so its stale work is
+        never sent (the platform shows a report without detail as reasoning)."""
+        owner = self._activity_owner
+        if owner is None or owner.aborted:
+            return None
+        return owner.activity.detail
+
+    async def observe_activity(
+        self,
+        scope: TurnScope,
+        *,
+        activity: AgentActivityDetail | None = None,
+        waiting: bool | None = None,
+    ) -> None:
+        """Record a turn's activity; never starts reporting or waits on it."""
+        before = self._visible_activity
+        scope.activity.apply(activity=activity, waiting=waiting)
+        if self._visible_activity != before:
+            self._working_reporter.refresh()
+
     def register_turn_task(self, scope: TurnScope, owner: TurnTask) -> None:
         """Bind completion to the same owner that delivers worker cancellation."""
         scope.bind(owner)
@@ -2158,7 +2183,6 @@ async def _abort_cycle(self, kind: ControlMode, msg_id: str | None) -> bool:
         # poisoned into permanently_failed before it's ever actually attempted.
         if msg_id:
             self._retry_tracker.discard_attempt(msg_id)
-        await self._clear_activity()
         if kind is ControlMode.INTERRUPT and msg_id:
             # Consume the message so the idle /next resync does not re-return it
             # (excludes-only-processed) and re-fire the cycle the user just
diff --git a/src/band/runtime/sessionconfig.py b/src/band/runtime/sessionconfig.py
index 9c9b73171..05e5a98e6 100644
--- a/src/band/runtime/sessionconfig.py
+++ b/src/band/runtime/sessionconfig.py
@@ -43,14 +43,13 @@ class SessionConfig(BaseSettings):
     # Queue traffic cannot postpone a check; active handlers remain serial.
     idle_resync_seconds: PositiveFiniteSeconds = 60.0
 
-    # --- Working-state (boolean "is the agent reasoning") reporting ---
-    # Kill-switch: disable all working-state reporting instantly if needed.
+    # Indicator and activity detail share this switch, including approval waits.
     enable_working_state: bool = True
     # Keep-alive cadence: how often working:true is refreshed while a cycle runs.
     # Must be < TTL/2 so a single missed/slow ping still lands inside the TTL.
     working_keep_alive_seconds: float = 3.0
-    # Per-POST deadline for each activity report (must be < cadence so a slow
-    # POST can't pile onto the next keep-alive tick).
+    # Total deadline for one report's POST and its compatibility fallback.
+    # Must be < cadence so a slow report cannot pile onto the next tick.
     working_request_timeout_seconds: int = 2
     # Optional upper bound on how long we keep asserting working:true for one
     # cycle. When exceeded we stop refreshing (platform TTL clears it); we never
diff --git a/src/band/runtime/tools/agent.py b/src/band/runtime/tools/agent.py
index 88d53634b..9067440a0 100644
--- a/src/band/runtime/tools/agent.py
+++ b/src/band/runtime/tools/agent.py
@@ -58,6 +58,7 @@
 from band.core.types import Capability, MessageType
 from band.core.validation import at_least_one_of
 from band.platform.posting import post_event, post_message
+from band.runtime.activity import ActivityObserver
 from band.runtime.capabilities import with_hub_room_contacts
 from band.runtime.context_serialization import context_item_to_dict
 from band.runtime.cycle import TurnScope
@@ -344,6 +345,7 @@ def __init__(
         self._agent_id = agent_id
         self._ctx: ExecutionContext | None = None
         self._scope: TurnScope | None = None
+        self._activity_observer: ActivityObserver | None = None
         self.turn = Turn()
 
     @property
@@ -384,6 +386,7 @@ def from_context(
         )
         tools._ctx = ctx
         tools._scope = ctx.current_scope if bind_scope else None
+        tools._activity_observer = ctx.activity_observer(tools._scope)
         tools.turn = Turn(posts_missing_reply=ctx.config.report_turn_failures_to_room)
         return tools
 
diff --git a/src/band/runtime/working_state.py b/src/band/runtime/working_state.py
index 4d842c7fe..540c46b9f 100644
--- a/src/band/runtime/working_state.py
+++ b/src/band/runtime/working_state.py
@@ -1,4 +1,4 @@
-"""Boolean working-state keep-alive reporter.
+"""Working-state keep-alive reporter.
 
 Owns the per-room "is the agent working" signal for a single reasoning cycle:
 sends ``working: true`` when a cycle starts, refreshes it on a keep-alive
@@ -14,12 +14,16 @@
 Contract: ``report`` MUST be time-bounded. The reporter awaits it (including the
 authoritative false-send during teardown), so an unbounded callback could wedge
 cycle teardown and cancellation propagation. ``BandLink.report_activity`` applies
-a per-POST timeout to satisfy this.
+one deadline per report, covering the POST and its compatibility resend.
 
-Cadence accuracy: the keep-alive sleeps ``keep_alive_seconds`` and *then* reports,
-so the effective interval between successful refreshes is
+The keep-alive task is the only sender of ``working: true`` after the first.
+``refresh()`` wakes it to report a changed activity now, so a caller never waits
+on the network and changes made while a report is in flight coalesce.
+
+Cadence accuracy: the keep-alive waits ``keep_alive_seconds`` (or a wake) and
+*then* reports, so the effective interval between successful refreshes is
 ``keep_alive_seconds + report-latency``. The platform-TTL headroom is preserved by
-the ``cadence < TTL/2`` guard on ``SessionConfig`` (with the per-POST timeout kept
+the ``cadence < TTL/2`` guard on ``SessionConfig`` (with the report deadline kept
 below the cadence), so even one missed/slow ping stays within the TTL window.
 """
 
@@ -28,6 +32,7 @@
 import asyncio
 import logging
 from collections.abc import Awaitable, Callable
+from contextlib import suppress
 
 logger = logging.getLogger(__name__)
 
@@ -40,7 +45,7 @@ def _retrieve_task_error(task: asyncio.Task[None]) -> None:
 
 
 class WorkingStateReporter:
-    """Manages the boolean working signal for one execution (agent + room)."""
+    """Manages the working signal and its activity for one execution (agent + room)."""
 
     def __init__(
         self,
@@ -63,6 +68,7 @@ def __init__(
         self._started_at: float | None = None
         self._task: asyncio.Task[None] | None = None
         self._false_send: asyncio.Future[None] | None = None
+        self._wake = asyncio.Event()
         # Serializes every report() call (true / keep-alive / false) so they can
         # never overlap on the wire — guarantees the platform sees them in order
         # (boolean payload carries no sequence number to disambiguate).
@@ -76,11 +82,18 @@ async def start(self) -> None:
             return
         await self._finish_false_send()
         self._active = True
+        self._wake.clear()
         self._started_at = asyncio.get_running_loop().time()
         await self._safe_report(True)
-        self._task = asyncio.create_task(
-            self._keep_alive(), name="working-state-keepalive"
-        )
+        # A false armed during the first report ends the cycle before it ran.
+        if self._active:
+            self._task = asyncio.create_task(
+                self._keep_alive(), name="working-state-keepalive"
+            )
+
+    def refresh(self) -> None:
+        """Report the current activity now instead of at the next keep-alive."""
+        self._wake.set()
 
     def defer_false_report(self) -> None:
         """Start ``working: false`` and return without waiting for it.
@@ -142,7 +155,7 @@ async def _finish_false_send(self) -> None:
             # platform clears the indicator immediately rather than via TTL.
             # Accepted gap: a *second* cancel landing during this await can abandon
             # it, leaving `send` detached. `send` is time-bounded (the report
-            # callback has a per-POST deadline) so it still completes on the loop;
+            # callback has a per-report deadline) so it still completes on the loop;
             # and the platform TTL backstops the clear regardless. Not handled
             # explicitly — the double-cancel teardown window is vanishingly rare.
             await send
@@ -152,9 +165,12 @@ async def _finish_false_send(self) -> None:
                 self._false_send = None
 
     async def _keep_alive(self) -> None:
-        """Re-send ``working: true`` on cadence until stopped or capped."""
+        """Re-send ``working: true`` on cadence or wake until stopped or capped."""
         while True:
-            await asyncio.sleep(self._keep_alive_seconds)
+            with suppress(TimeoutError):
+                async with asyncio.timeout(self._keep_alive_seconds):
+                    await self._wake.wait()
+            self._wake.clear()
             if not self._active:
                 return
             if self._cap_reached():
diff --git a/tests/activitysupport.py b/tests/activitysupport.py
new file mode 100644
index 000000000..a046463dd
--- /dev/null
+++ b/tests/activitysupport.py
@@ -0,0 +1,47 @@
+"""Activity report bodies and per-test isolation of the detail negotiation."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+import band.platform.message_lifecycle as lifecycle
+from band.platform.activity import (
+    API_VERSION_HEADER,
+    MINIMUM_ACTIVITY_VERSION,
+    ActivitySupport,
+)
+from band.runtime.activity import MEMORY, PERMISSION, REASONING, TOOL_CALL
+
+SUPPORTED_VERSION = ".".join(map(str, MINIMUM_ACTIVITY_VERSION))
+
+BOOLEAN_BODY = {"working": True}
+IDLE_BODY = {"working": False}
+
+
+def detail_body(kind: str, name: str | None = None) -> dict[str, Any]:
+    activity = {"kind": kind} if name is None else {"kind": kind, "name": name}
+    return {**BOOLEAN_BODY, "activity": activity}
+
+
+def tool_body(name: str) -> dict[str, Any]:
+    return detail_body(TOOL_CALL, name)
+
+
+REASONING_BODY = detail_body(REASONING.kind)
+PERMISSION_BODY = detail_body(PERMISSION.kind)
+MEMORY_BODY = detail_body(MEMORY.kind)
+DETAIL_BODY = tool_body("Bash")
+DETAIL = DETAIL_BODY["activity"]
+
+
+def install_activity_support(
+    monkeypatch: pytest.MonkeyPatch, *, learned: bool
+) -> ActivitySupport:
+    """Give the lifecycle a fresh negotiation, optionally already supported."""
+    support = ActivitySupport()
+    if learned:
+        support.observe({API_VERSION_HEADER: SUPPORTED_VERSION})
+    monkeypatch.setattr(lifecycle, "activity_support", support)
+    return support
diff --git a/tests/e2e/baseline/toolkit/control.py b/tests/e2e/baseline/toolkit/control.py
index 70e5c1764..2b5087d92 100644
--- a/tests/e2e/baseline/toolkit/control.py
+++ b/tests/e2e/baseline/toolkit/control.py
@@ -7,6 +7,8 @@
 from collections.abc import AsyncGenerator, Callable
 from contextlib import asynccontextmanager
 
+from band_rest.types.agent_activity_detail import AgentActivityDetail
+
 import band.runtime
 from band.client.streaming import (
     AgentControlPayload,
@@ -195,6 +197,7 @@ async def report_activity(
         room_id: str,
         working: bool,
         *,
+        activity: AgentActivityDetail | None = None,
         timeout_seconds: int = 2,
         on_stopped: Callable[[], None] | None = None,
     ) -> bool:
@@ -205,7 +208,11 @@ def observe_stop() -> None:
                 on_stopped()
 
         accepted = await super().report_activity(
-            room_id, working, timeout_seconds=timeout_seconds, on_stopped=observe_stop
+            room_id,
+            working,
+            activity=activity,
+            timeout_seconds=timeout_seconds,
+            on_stopped=observe_stop,
         )
         if accepted and not working and room_id == self.control_room_id:
             self.control.activity_idle.set()
diff --git a/tests/framework_conformance/test_collected_coverage.py b/tests/framework_conformance/test_collected_coverage.py
index 8c416952c..1e42e438f 100644
--- a/tests/framework_conformance/test_collected_coverage.py
+++ b/tests/framework_conformance/test_collected_coverage.py
@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import sys
 from pathlib import Path
 
+import pytest
+
 from tests.framework_conformance.venv_job_coverage import (
     collected_nodes,
     selected_nodes,
@@ -12,7 +15,7 @@
 
 class TestCollectedCoverage:
     def test_keyword_filter_does_not_cover_an_explicitly_named_file(
-        self, tmp_path: Path
+        self, tmp_path: Path, pytester: pytest.Pytester
     ) -> None:
         test = tmp_path / "test_selection.py"
         test.write_text(
@@ -20,12 +23,12 @@ def test_keyword_filter_does_not_cover_an_explicitly_named_file(
             "    def test_retained(self): pass\n"
             "    def test_omitted(self): pass\n"
         )
-        full = selected_nodes([str(test)])
-        selected = selected_nodes([str(test), "-k", "retained"])
+        full = selected_nodes(pytester, [str(test)])
+        selected = selected_nodes(pytester, [str(test), "-k", "retained"])
         assert {node.rsplit("::", 1)[1] for node in full - selected} == {"test_omitted"}
 
     def test_framework_binding_does_not_depend_on_the_test_name(
-        self, tmp_path: Path
+        self, tmp_path: Path, pytester: pytest.Pytester
     ) -> None:
         test = tmp_path / "test_bindings.py"
         test.write_text(
@@ -34,8 +37,35 @@ def test_framework_binding_does_not_depend_on_the_test_name(
             '    @pytest.mark.parametrize("framework_id", ["strands", "pydantic_ai"])\n'
             "    def test_behavior(self, framework_id): pass\n"
         )
-        nodes = collected_nodes([str(test)])
+        nodes = collected_nodes(pytester, [str(test)])
         assert {node.frameworks for node in nodes} == {
             frozenset({"strands"}),
             frozenset({"pydantic_ai"}),
         }
+
+    def test_ignored_file_does_not_count_as_covered(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        pytester.makeini("[pytest]")
+        retained = pytester.makepyfile(test_retained="def test_behavior(): pass")
+        omitted = pytester.makepyfile(test_omitted="def test_behavior(): pass")
+        full = selected_nodes(pytester, [str(pytester.path)])
+        selected = selected_nodes(
+            pytester, [str(pytester.path), "--ignore", str(omitted)]
+        )
+        assert {node.split("::", 1)[0] for node in full - selected} == {omitted.name}
+        assert {node.split("::", 1)[0] for node in selected} == {retained.name}
+
+    def test_collection_restores_imported_modules(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        test = pytester.makepyfile(test_isolated_import="def test_behavior(): pass")
+        assert selected_nodes(pytester, [str(test)])
+        assert test.stem not in sys.modules
+
+    def test_collection_errors_fail_the_coverage_check(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        test = pytester.makepyfile('raise RuntimeError("collection failed")')
+        with pytest.raises(AssertionError):
+            collected_nodes(pytester, [str(test)])
diff --git a/tests/framework_conformance/test_strands_nooa_job_coverage.py b/tests/framework_conformance/test_strands_nooa_job_coverage.py
index d639dd29e..71982e606 100644
--- a/tests/framework_conformance/test_strands_nooa_job_coverage.py
+++ b/tests/framework_conformance/test_strands_nooa_job_coverage.py
@@ -32,7 +32,9 @@ def test_expected_configs_are_present(self, framework_environment: None) -> None
         assert expected <= {config.framework_id for config in ADAPTER_CONFIGS}
         assert expected <= {config.framework_id for config in CONVERTER_CONFIGS}
 
-    def test_job_selects_every_required_node(self, framework_environment: None) -> None:
+    def test_job_selects_every_required_node(
+        self, framework_environment: None, pytester: pytest.Pytester
+    ) -> None:
         needed = coverage.tests_needing_venv(
             coverage.needs_venv_pattern(frozenset({"strands", "nooa"}))
         )
@@ -41,7 +43,7 @@ def test_job_selects_every_required_node(self, framework_environment: None) -> N
         arguments = coverage.pytest_arguments(
             coverage.job_command("test-strands-nooa", "Run Strands and NOOA tests")
         )
-        selected = coverage.selected_nodes(arguments)
+        selected = coverage.selected_nodes(pytester, arguments)
         # Directory collection honors NOOA's supported-platform conftest gate.
         nooa_tests = REPO_ROOT / "tests/adapters/nooa"
         targets = {
@@ -49,7 +51,7 @@ def test_job_selects_every_required_node(self, framework_environment: None) -> N
             for path in needed
         }
         baseline = coverage.collected_nodes(
-            [*sorted(targets), "tests/framework_conformance/"]
+            pytester, [*sorted(targets), "tests/framework_conformance/"]
         )
         needed_files = {path.as_posix() for path in needed}
         required = {
diff --git a/tests/framework_conformance/venv_job_coverage.py b/tests/framework_conformance/venv_job_coverage.py
index e5a59ee79..bc8ef1e66 100644
--- a/tests/framework_conformance/venv_job_coverage.py
+++ b/tests/framework_conformance/venv_job_coverage.py
@@ -3,16 +3,12 @@
 from __future__ import annotations
 
 import ast
-import json
 import re
 import shlex
-import subprocess
-import sys
-import tempfile
 import tomllib
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 import pytest
 import yaml
@@ -135,34 +131,26 @@ class CollectedNode:
     skipped: bool
 
 
-def collected_nodes(arguments: list[str]) -> tuple[CollectedNode, ...]:
+def collected_nodes(
+    pytester: pytest.Pytester, arguments: list[str]
+) -> tuple[CollectedNode, ...]:
     """Collect nodes and framework bindings after pytest applies the job selectors."""
-    with tempfile.TemporaryDirectory() as directory:
-        output = Path(directory) / "nodes.json"
-        result = subprocess.run(
-            [
-                sys.executable,
-                "-m",
-                "tests.framework_conformance.venv_job_coverage",
-                str(output),
-                *arguments,
-            ],
-            cwd=REPO_ROOT,
-            capture_output=True,
-            text=True,
-            check=False,
-        )
-        if result.returncode != 0:
-            raise AssertionError(result.stdout + result.stderr)
-        return tuple(
-            CollectedNode(row["nodeid"], frozenset(row["frameworks"]), row["skipped"])
-            for row in json.loads(output.read_text(encoding="utf-8"))
+    with pytest.MonkeyPatch.context() as monkeypatch:
+        monkeypatch.chdir(REPO_ROOT)
+        recorder = pytester.inline_run(*arguments, "--collect-only", "--no-cov")
+    assert recorder.ret == pytest.ExitCode.OK
+    session = cast(pytest.Session, recorder.getcall("pytest_collection_finish").session)
+    return tuple(
+        CollectedNode(
+            item.nodeid, frozenset(framework_bindings(item)), is_skipped(item)
         )
+        for item in session.items
+    )
 
 
-def selected_nodes(arguments: list[str]) -> set[str]:
+def selected_nodes(pytester: pytest.Pytester, arguments: list[str]) -> set[str]:
     """The exact node IDs surviving a job's pytest selectors."""
-    return {node.nodeid for node in collected_nodes(arguments)}
+    return {node.nodeid for node in collected_nodes(pytester, arguments)}
 
 
 def framework_bindings(item: pytest.Item) -> set[str]:
@@ -175,38 +163,13 @@ def framework_bindings(item: pytest.Item) -> set[str]:
     }
 
 
-def collection_row(item: pytest.Item) -> dict[str, object]:
-    skipped = item.get_closest_marker("skip") is not None or any(
+def is_skipped(item: pytest.Item) -> bool:
+    return item.get_closest_marker("skip") is not None or any(
         marker.args and marker.args[0] is True for marker in item.iter_markers("skipif")
     )
-    return {
-        "nodeid": item.nodeid,
-        "frameworks": sorted(framework_bindings(item)),
-        "skipped": skipped,
-    }
 
 
 def pytest_arguments(command: str) -> list[str]:
     """Read selectors after the workflow's pytest executable."""
     tokens = shlex.split(command)
     return tokens[tokens.index("pytest") + 1 :]
-
-
-class NodeCollector:
-    def __init__(self, output: Path) -> None:
-        self.output = output
-
-    def pytest_collection_finish(self, session: pytest.Session) -> None:
-        self.output.write_text(
-            json.dumps([collection_row(item) for item in session.items]),
-            encoding="utf-8",
-        )
-
-
-if __name__ == "__main__":
-    raise SystemExit(
-        pytest.main(
-            [*sys.argv[2:], "--collect-only", "--no-cov"],
-            plugins=[NodeCollector(Path(sys.argv[1]))],
-        )
-    )
diff --git a/tests/platform/activity/conftest.py b/tests/platform/activity/conftest.py
new file mode 100644
index 000000000..efe8f3861
--- /dev/null
+++ b/tests/platform/activity/conftest.py
@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+
+import pytest
+from band_rest import AgentActivityDetail
+
+from band.platform.activity import ActivitySupport
+from band.platform.link import BandLink
+from tests.activitysupport import DETAIL, install_activity_support
+from tests.platform.activity.kit import ActivityPeer
+from tests.runtime.helpers import rest_client_over
+
+
+@pytest.fixture(autouse=True)
+def support(monkeypatch: pytest.MonkeyPatch) -> ActivitySupport:
+    return install_activity_support(monkeypatch, learned=False)
+
+
+@pytest.fixture
+def activity_peer() -> ActivityPeer:
+    return ActivityPeer()
+
+
+@pytest.fixture
+def detail() -> AgentActivityDetail:
+    return AgentActivityDetail(**DETAIL)
+
+
+@pytest.fixture
+async def link(activity_peer: ActivityPeer) -> AsyncIterator[BandLink]:
+    """A link on the SDK's default (SaaS) ``rest_url``."""
+    async with rest_client_over(activity_peer.respond) as rest:
+        link = BandLink(agent_id="agent", api_key="test-key")
+        link.rest = rest
+        yield link
+
+
+@pytest.fixture
+def second_link(link: BandLink) -> BandLink:
+    other = BandLink(agent_id="second-agent", api_key="test-key")
+    other.rest = link.rest
+    return other
+
+
+@pytest.fixture
+def learned(monkeypatch: pytest.MonkeyPatch) -> ActivitySupport:
+    """A process that already saw a supporting ``X-API-Version``."""
+    return install_activity_support(monkeypatch, learned=True)
diff --git a/tests/platform/activity/kit.py b/tests/platform/activity/kit.py
new file mode 100644
index 000000000..51b1a14cc
--- /dev/null
+++ b/tests/platform/activity/kit.py
@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from collections import deque
+from typing import Any
+
+import httpx
+import pytest
+
+from band.platform.activity import API_VERSION_HEADER
+from band.platform.message_lifecycle import logger as lifecycle_logger
+from tests.activitysupport import SUPPORTED_VERSION
+
+
+def lifecycle_records(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]:
+    """Records of the delivery logger alone, not the version negotiation's."""
+    return [r for r in caplog.records if r.name == lifecycle_logger.name]
+
+
+class ActivityPeer:
+    """The activity endpoint, advertising ``version`` on every response.
+
+    ``post_latency`` maps a 1-based POST number to the seconds it takes to answer.
+    """
+
+    def __init__(self) -> None:
+        self.version: str | None = SUPPORTED_VERSION
+        self.reports: list[dict[str, Any]] = []
+        self.requests: list[httpx.Request] = []
+        self.responses: deque[httpx.Response] = deque()
+        self.stall_post_number: int | None = None
+        self.post_latency: dict[int, float] = {}
+
+    async def respond(self, request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        self.requests.append(request)
+        self.reports.append(body)
+        if latency := self.post_latency.get(len(self.reports)):
+            await asyncio.sleep(latency)
+        if len(self.reports) == self.stall_post_number:
+            await asyncio.Event().wait()
+        response = (
+            self.responses.popleft()
+            if self.responses
+            else httpx.Response(200, json={"data": {"working": body["working"]}})
+        )
+        if self.version is not None:
+            response.headers[API_VERSION_HEADER] = self.version
+        return response
diff --git a/tests/platform/activity/test_control_link.py b/tests/platform/activity/test_control_link.py
new file mode 100644
index 000000000..00e8d78c6
--- /dev/null
+++ b/tests/platform/activity/test_control_link.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import pytest
+from band_rest import AgentActivityDetail
+
+from band.platform.link import BandLink
+from tests.activitysupport import DETAIL_BODY
+from tests.e2e.baseline.toolkit.control import ControlLink, ControlRuntime
+from tests.platform.activity.kit import ActivityPeer
+
+
+@pytest.mark.usefixtures("learned")
+class TestControlLinkActivity:
+    async def test_baseline_control_link_forwards_activity_detail(
+        self, link: BandLink, activity_peer: ActivityPeer, detail: AgentActivityDetail
+    ) -> None:
+        control_link = ControlLink(agent_id="agent", api_key="test-key")
+        control_link.rest = link.rest
+        control_link.control = ControlRuntime()
+        control_link.control_room_id = "room"
+        assert await control_link.report_activity("room", True, activity=detail)
+        assert activity_peer.reports == [DETAIL_BODY]
diff --git a/tests/platform/activity/test_deadlines.py b/tests/platform/activity/test_deadlines.py
new file mode 100644
index 000000000..5c36e6f3c
--- /dev/null
+++ b/tests/platform/activity/test_deadlines.py
@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+import pytest
+from band_rest import AgentActivityDetail
+
+from band.platform.link import BandLink
+from tests.activitysupport import BOOLEAN_BODY, DETAIL_BODY
+from tests.platform.activity.kit import ActivityPeer
+
+DEADLINE_SECONDS = 2
+# Under the deadline alone, but the deadline cannot cover it plus a full resend.
+REJECTION_LATENCY_SECONDS = 1.5
+
+
+@pytest.fixture
+def stalled_resend(activity_peer: ActivityPeer) -> ActivityPeer:
+    """The detail POST is slowly rejected and its detail-free resend never answers."""
+    activity_peer.post_latency[1] = REJECTION_LATENCY_SECONDS
+    activity_peer.responses.append(httpx.Response(422))
+    activity_peer.stall_post_number = 2
+    return activity_peer
+
+
+@pytest.mark.looptime
+@pytest.mark.usefixtures("learned")
+class TestActivityDeadline:
+    async def test_resend_cannot_extend_the_report_deadline(
+        self, link: BandLink, stalled_resend: ActivityPeer, detail: AgentActivityDetail
+    ) -> None:
+        started = asyncio.get_running_loop().time()
+        assert not await link.report_activity(
+            "room", True, activity=detail, timeout_seconds=DEADLINE_SECONDS
+        )
+        assert asyncio.get_running_loop().time() - started == DEADLINE_SECONDS
+        assert stalled_resend.reports == [
+            DETAIL_BODY,
+            BOOLEAN_BODY,
+        ]
+
+    async def test_downgrade_survives_a_timed_out_resend(
+        self, link: BandLink, stalled_resend: ActivityPeer, detail: AgentActivityDetail
+    ) -> None:
+        await link.report_activity(
+            "room", True, activity=detail, timeout_seconds=DEADLINE_SECONDS
+        )
+        stalled_resend.stall_post_number = None
+        assert await link.report_activity("next", True, activity=detail)
+        assert stalled_resend.reports[-1] == BOOLEAN_BODY
diff --git a/tests/platform/activity/test_fallback.py b/tests/platform/activity/test_fallback.py
new file mode 100644
index 000000000..dfb20b8d6
--- /dev/null
+++ b/tests/platform/activity/test_fallback.py
@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+import httpx
+import pytest
+from band_rest import AgentActivityDetail
+
+from band.platform.activity import ActivitySupport
+from band.platform.link import BandLink
+from tests.activitysupport import BOOLEAN_BODY, DETAIL_BODY
+from tests.platform.activity.kit import ActivityPeer
+
+
+@pytest.mark.usefixtures("learned")
+class TestActivityFallback:
+    @pytest.mark.parametrize(
+        "body", [b'{"errors":{"activity":["unknown"]}}', b"not json", b""]
+    )
+    async def test_422_retries_without_detail_and_latches_across_links(
+        self,
+        link: BandLink,
+        second_link: BandLink,
+        activity_peer: ActivityPeer,
+        detail: AgentActivityDetail,
+        body: bytes,
+    ) -> None:
+        activity_peer.responses.append(httpx.Response(422, content=body))
+        assert await link.report_activity("first", True, activity=detail)
+        assert await second_link.report_activity("second", True, activity=detail)
+        assert activity_peer.reports == [
+            DETAIL_BODY,
+            BOOLEAN_BODY,
+            BOOLEAN_BODY,
+        ]
+
+    @pytest.mark.parametrize(
+        ("working", "with_detail"),
+        [(True, False), (False, True)],
+        ids=["no-detail", "idle"],
+    )
+    async def test_422_without_detail_neither_retries_nor_latches(
+        self,
+        link: BandLink,
+        activity_peer: ActivityPeer,
+        learned: ActivitySupport,
+        detail: AgentActivityDetail,
+        working: bool,
+        with_detail: bool,
+    ) -> None:
+        activity_peer.responses.append(httpx.Response(422))
+        activity = detail if with_detail else None
+        assert not await link.report_activity("room", working, activity=activity)
+        assert len(activity_peer.reports) == 1
+        assert learned.supported
+
+    @pytest.mark.parametrize("status", [400, 403, 500])
+    async def test_other_errors_do_not_retry_or_disable(
+        self,
+        link: BandLink,
+        activity_peer: ActivityPeer,
+        detail: AgentActivityDetail,
+        status: int,
+    ) -> None:
+        activity_peer.responses.append(
+            httpx.Response(status, json={"error": "failure"})
+        )
+        assert not await link.report_activity("room", True, activity=detail)
+        assert await link.report_activity("room", True, activity=detail)
+        assert activity_peer.reports == [DETAIL_BODY] * 2
+
+    async def test_fallback_echoed_false_keeps_stop_notification(
+        self, link: BandLink, activity_peer: ActivityPeer, detail: AgentActivityDetail
+    ) -> None:
+        activity_peer.responses.extend(
+            [
+                httpx.Response(422),
+                httpx.Response(200, json={"data": {"working": False}}),
+            ]
+        )
+        stopped = Mock()
+        assert await link.report_activity(
+            "room", True, activity=detail, on_stopped=stopped
+        )
+        stopped.assert_called_once_with()
+        assert activity_peer.reports[-1] == BOOLEAN_BODY
diff --git a/tests/platform/activity/test_reporting.py b/tests/platform/activity/test_reporting.py
new file mode 100644
index 000000000..0c115d798
--- /dev/null
+++ b/tests/platform/activity/test_reporting.py
@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import logging
+
+import httpx
+import pytest
+
+from band.platform.link import BandLink
+from tests.platform.activity.kit import ActivityPeer, lifecycle_records
+
+
+class TestActivityReporting:
+    @pytest.mark.parametrize("working", [True, False])
+    async def test_reports_the_boolean_state_to_the_room(
+        self, link: BandLink, activity_peer: ActivityPeer, working: bool
+    ) -> None:
+        assert await link.report_activity("room-1", working)
+        assert activity_peer.reports == [{"working": working}]
+        assert activity_peer.requests[-1].url.path.endswith("/chats/room-1/activity")
+
+    async def test_post_is_bounded_and_never_retried(
+        self, link: BandLink, activity_peer: ActivityPeer
+    ) -> None:
+        activity_peer.responses.append(httpx.Response(500))
+        assert not await link.report_activity("room-1", True, timeout_seconds=2)
+        [request] = activity_peer.requests
+        assert request.extensions["timeout"]["read"] == 2
+
+    async def test_repeated_failures_warn_once_then_recover(
+        self,
+        link: BandLink,
+        activity_peer: ActivityPeer,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        activity_peer.responses.extend([httpx.Response(500), httpx.Response(500)])
+        with caplog.at_level(logging.INFO, logger="band.platform.message_lifecycle"):
+            assert not await link.report_activity("room-1", True)
+            assert not await link.report_activity("room-1", True)
+            assert await link.report_activity("room-1", True)
+        records = lifecycle_records(caplog)
+        assert [record.levelno for record in records] == [
+            logging.WARNING,
+            logging.INFO,
+        ]
+        assert "recovered" in records[-1].getMessage()
diff --git a/tests/platform/activity/test_versions.py b/tests/platform/activity/test_versions.py
new file mode 100644
index 000000000..224e8bf12
--- /dev/null
+++ b/tests/platform/activity/test_versions.py
@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import pytest
+from band_rest import AgentActivityDetail
+
+from band.platform.link import BandLink
+from tests.activitysupport import (
+    BOOLEAN_BODY,
+    DETAIL_BODY,
+    IDLE_BODY,
+    SUPPORTED_VERSION,
+    detail_body,
+)
+from tests.platform.activity.kit import ActivityPeer
+
+
+class TestActivityVersion:
+    @pytest.mark.parametrize(
+        ("version", "supported"),
+        [
+            ("1.24.99", False),
+            pytest.param(SUPPORTED_VERSION, True, id="minimum-version-on-saas-default"),
+            ("1.26.10", True),
+            ("2.0.0", True),
+            ("1.25.0-beta", False),
+            ("broken", False),
+            ("1.25", False),
+            ("2.0", False),
+            (None, False),
+        ],
+    )
+    async def test_advertised_version_controls_later_detail(
+        self,
+        link: BandLink,
+        activity_peer: ActivityPeer,
+        detail: AgentActivityDetail,
+        version: str | None,
+        supported: bool,
+    ) -> None:
+        activity_peer.version = version
+        assert await link.report_activity("room", True, activity=detail)
+        assert await link.report_activity("room", True, activity=detail)
+        later = DETAIL_BODY if supported else BOOLEAN_BODY
+        assert activity_peer.reports == [BOOLEAN_BODY, later]
+        assert all(
+            request.url.path.endswith("/activity") for request in activity_peer.requests
+        )
+
+    async def test_learned_support_is_shared_across_links(
+        self,
+        link: BandLink,
+        second_link: BandLink,
+        activity_peer: ActivityPeer,
+        detail: AgentActivityDetail,
+    ) -> None:
+        assert await link.report_activity("first", True)
+        assert await second_link.report_activity("second", True, activity=detail)
+        assert activity_peer.reports[-1] == DETAIL_BODY
+
+    @pytest.mark.parametrize("kind", ["reasoning", "memory", "permission"])
+    async def test_non_tool_detail_omits_name_and_false_omits_detail(
+        self, link: BandLink, activity_peer: ActivityPeer, kind: str
+    ) -> None:
+        detail = AgentActivityDetail(kind=kind)
+        assert await link.report_activity("room", True)
+        assert await link.report_activity("room", True, activity=detail)
+        assert await link.report_activity("room", False, activity=detail)
+        assert activity_peer.reports[1:] == [
+            detail_body(kind),
+            IDLE_BODY,
+        ]
diff --git a/tests/platform/test_link.py b/tests/platform/test_link.py
index d6401a71d..8a17eeb79 100644
--- a/tests/platform/test_link.py
+++ b/tests/platform/test_link.py
@@ -3,16 +3,9 @@
 from __future__ import annotations
 
 import asyncio
-import logging
 from unittest.mock import ANY, AsyncMock, MagicMock, create_autospec, patch
 
 import pytest
-from band_rest import (
-    AsyncRestClient,
-    NotFoundError,
-    UnauthorizedError,
-    UnprocessableEntityError,
-)
 from band_rest.core.api_error import ApiError
 from band_rest.types.chat_message_metadata import ChatMessageMetadata
 from band_sdk_core import AgentTopicStatus, DeadReason, SessionState, chat_room_topic
@@ -1346,135 +1339,3 @@ async def test_returns_empty_list_on_failure(self):
         messages = await link.get_stale_processing_messages("room-1")
 
         assert messages == []
-
-
-class TestReportActivity:
-    """Tests for BandLink.report_activity (boolean working-state reporting)."""
-
-    @pytest.mark.asyncio
-    async def test_reports_working_true(self):
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock()
-
-        result = await link.report_activity("room-1", True)
-
-        assert result is True
-        call = link.rest.agent_api_activity.report_agent_chat_activity
-        call.assert_awaited_once()
-        assert call.call_args.kwargs["chat_id"] == "room-1"
-        assert call.call_args.kwargs["working"] is True
-
-    @pytest.mark.asyncio
-    async def test_passes_per_post_timeout_and_no_retries(self):
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock()
-
-        await link.report_activity("room-1", True, timeout_seconds=2)
-
-        opts = link.rest.agent_api_activity.report_agent_chat_activity.call_args.kwargs[
-            "request_options"
-        ]
-        assert opts["timeout_in_seconds"] == 2
-        assert opts["max_retries"] == 0
-
-    @pytest.mark.asyncio
-    async def test_reports_working_false(self):
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock()
-
-        result = await link.report_activity("room-1", False)
-
-        assert result is True
-        call = link.rest.agent_api_activity.report_agent_chat_activity
-        assert call.call_args.kwargs["working"] is False
-
-    @pytest.mark.asyncio
-    async def test_returns_false_on_not_found(self):
-
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock(
-            side_effect=NotFoundError(headers={}, body="no active execution")
-        )
-
-        result = await link.report_activity("room-1", True)
-
-        assert result is False
-
-    @pytest.mark.asyncio
-    async def test_returns_false_on_unauthorized(self):
-
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock(
-            side_effect=UnauthorizedError(headers={}, body="bad key")
-        )
-
-        result = await link.report_activity("room-1", True)
-
-        assert result is False
-
-    @pytest.mark.asyncio
-    async def test_returns_false_on_unprocessable_entity(self):
-
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock(
-            side_effect=UnprocessableEntityError(headers={}, body="bad uuid")
-        )
-
-        result = await link.report_activity("room-1", True)
-
-        assert result is False
-
-    @pytest.mark.asyncio
-    async def test_returns_false_on_network_error(self):
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock(
-            side_effect=Exception("network down")
-        )
-
-        result = await link.report_activity("room-1", True)
-
-        assert result is False
-
-    def test_real_client_exposes_activity_method(self):
-        """Guard: the real REST client must actually expose the activity method.
-
-        The AsyncMock-based tests above auto-fabricate attributes, so they would
-        stay green even if `agent_api_activity.report_agent_chat_activity`
-        disappeared or was renamed in a band_rest bump. This test pins the
-        real wire contract: instantiate the real client and assert the method
-        exists and is callable.
-        """
-
-        client = AsyncRestClient(api_key="test-key", base_url="https://test.com")
-        method = getattr(client.agent_api_activity, "report_agent_chat_activity", None)
-        assert callable(method)
-
-    @pytest.mark.asyncio
-    async def test_repeated_failures_warn_once_then_recover(self, caplog):
-
-        link = BandLink(agent_id="agent-123", api_key="test-key")
-        link.rest = MagicMock()
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock(
-            side_effect=Exception("down")
-        )
-
-        with caplog.at_level(logging.WARNING, logger="band.platform.message_lifecycle"):
-            assert await link.report_activity("room-1", True) is False
-            assert await link.report_activity("room-1", True) is False
-
-        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
-        assert len(warnings) == 1  # debounced: only the first failure warns
-
-        # Recovery logs once at INFO and re-arms the warning.
-        link.rest.agent_api_activity.report_agent_chat_activity = AsyncMock()
-        with caplog.at_level(logging.INFO, logger="band.platform.message_lifecycle"):
-            caplog.clear()
-            assert await link.report_activity("room-1", True) is True
-        assert any("recovered" in r.message.lower() for r in caplog.records)
diff --git a/tests/runtime/activity/test_turndetail.py b/tests/runtime/activity/test_turndetail.py
new file mode 100644
index 000000000..bb87d8f39
--- /dev/null
+++ b/tests/runtime/activity/test_turndetail.py
@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from band_rest import AgentActivityDetail
+
+from band.runtime.activity import UNKNOWN_ACTIVITY_TOOL_NAME, TurnActivity
+
+
+class TestTurnActivity:
+    def test_a_turn_starts_at_reasoning(self) -> None:
+        assert TurnActivity().detail == AgentActivityDetail(kind="reasoning")
+
+    def test_only_the_kind_and_tool_name_are_kept(self) -> None:
+        activity = TurnActivity()
+        activity.apply(
+            activity=AgentActivityDetail(
+                kind="tool_call", name="Bash", arguments="private"
+            )
+        )
+        assert activity.detail.model_dump(exclude_unset=True) == {
+            "kind": "tool_call",
+            "name": "Bash",
+        }
+        activity.apply(activity=AgentActivityDetail(kind="memory", name="private"))
+        assert activity.detail.model_dump(exclude_unset=True) == {"kind": "memory"}
+
+    def test_a_nameless_tool_is_reported_as_unknown(self) -> None:
+        activity = TurnActivity()
+        activity.apply(activity=AgentActivityDetail(kind="tool_call", name=""))
+        assert activity.detail == AgentActivityDetail(
+            kind="tool_call", name=UNKNOWN_ACTIVITY_TOOL_NAME
+        )
+
+    def test_a_person_wait_overrides_progress_without_losing_it(self) -> None:
+        activity = TurnActivity()
+        activity.apply(activity=AgentActivityDetail(kind="tool_call", name="Read"))
+        activity.apply(waiting=True)
+        activity.apply(activity=AgentActivityDetail(kind="memory"))
+        assert activity.detail == AgentActivityDetail(kind="permission")
+        activity.apply(waiting=False)
+        assert activity.detail == AgentActivityDetail(kind="memory")
diff --git a/tests/runtime/activityownership/conftest.py b/tests/runtime/activityownership/conftest.py
index 2663643c5..20f44a74e 100644
--- a/tests/runtime/activityownership/conftest.py
+++ b/tests/runtime/activityownership/conftest.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
 from unittest.mock import AsyncMock
 
 import pytest
@@ -10,6 +10,7 @@
 from band.runtime.execution import ExecutionContext
 from band.runtime.runtime import AgentRuntime
 from band.runtime.sessionconfig import SessionConfig
+from band.runtime.tools.agent import AgentTools
 from tests.runtime.activityownership.kit import ActivityPlatform, IdleReportGate
 from tests.runtime.helpers import (
     AGENT_ID,
@@ -19,8 +20,12 @@
 
 
 @pytest.fixture
-async def activity_platform() -> AsyncIterator[ActivityPlatform]:
-    yield ActivityPlatform()
+def activity_platform() -> Iterator[ActivityPlatform]:
+    platform = ActivityPlatform()
+    yield platform
+    assert platform.late_changes == [], (
+        f"visible changes waited for a keep-alive (body, seconds): {platform.late_changes}"
+    )
 
 
 @pytest.fixture
@@ -37,6 +42,7 @@ async def execution(
 
         def create(
             handler: Callable[[ExecutionContext, PlatformEvent], Awaitable[None]],
+            hub_room_id: str | None = None,
             *,
             on_message_control: Callable[
                 [ExecutionContext, PlatformEvent], Awaitable[bool]
@@ -54,6 +60,7 @@ def create(
                 link,
                 handler,
                 agent_id=AGENT_ID,
+                hub_room_id=hub_room_id,
                 on_platform_stop=runtime._on_platform_stop,
                 on_message_control=on_message_control,
                 config=SessionConfig(enable_context_hydration=False, **settings),
@@ -63,4 +70,12 @@ def create(
 
         yield create
         for ctx in contexts:
+            idle_report_gate.peer.activity_release.set()
             await ctx.stop()
+
+
+@pytest.fixture
+async def bare_tools(activity_platform: ActivityPlatform) -> AsyncIterator[AgentTools]:
+    """Real tools built outside an execution, as adapters' unit tests build them."""
+    async with rest_client_over(activity_platform.respond) as rest:
+        yield AgentTools(ROOM_ID, rest)
diff --git a/tests/runtime/activityownership/kit.py b/tests/runtime/activityownership/kit.py
index 75284e34b..8ebb39513 100644
--- a/tests/runtime/activityownership/kit.py
+++ b/tests/runtime/activityownership/kit.py
@@ -2,13 +2,25 @@
 
 import asyncio
 import json
+from collections import deque
+from collections.abc import Awaitable
+from typing import Any, Literal
 
 import httpx
 
 from band.core.turn_lifecycle import TurnTask, bind_turn_task
+from band.platform.activity import API_VERSION_HEADER
 from band.platform.event import PlatformEvent
+from band.runtime.activity import (
+    end_tool_activity,
+    report_memory_activity,
+    report_tool_activity,
+    set_permission_wait,
+)
 from band.runtime.execution import ExecutionContext
+from band.runtime.sessionconfig import SessionConfig
 from band.runtime.tools.agent import AgentTools
+from tests.activitysupport import SUPPORTED_VERSION
 from tests.runtime.helpers import ALICE, LifecyclePlatform, is_message_post, page
 
 
@@ -16,21 +28,87 @@ def message_content(request: httpx.Request) -> str:
     return json.loads(request.content)["message"]["content"]
 
 
+KEEP_ALIVE_SECONDS = SessionConfig().working_keep_alive_seconds
+
+
+async def past_a_keep_alive() -> None:
+    """Let a keep-alive fall due, so a wrongly woken reporter would have sent."""
+    await asyncio.sleep(2 * KEEP_ALIVE_SECONDS)
+
+
+async def report_every_kind(tools: object) -> None:
+    await report_tool_activity(tools, "Bash")
+    await set_permission_wait(tools, True)
+    await report_memory_activity(tools)
+    await set_permission_wait(tools, False)
+    await end_tool_activity(tools)
+
+
 class ActivityPlatform(LifecyclePlatform):
-    """A running room whose participants are listed and whose messages land."""
+    """Activity posts advertise a detail-supporting version; queued responses
+    answer them before the default echo."""
 
     def __init__(self) -> None:
         super().__init__()
         self.room_messages: list[str] = []
+        self.activity_responses: deque[httpx.Response] = deque()
+        self.activity_bodies: list[dict[str, Any]] = []
+        self.hold_next_activity = False
+        self.activity_entered = asyncio.Event()
+        self.activity_release = asyncio.Event()
+        self.late_changes: list[tuple[dict[str, Any], float]] = []
+        self._reported = asyncio.Condition()
 
     async def respond(self, request: httpx.Request) -> httpx.Response:
-        return self.answer(request)
+        response = self.answer(request)
+        if request.url.path.endswith("/activity"):
+            if self.activity_responses:
+                response = self.activity_responses.popleft()
+            response.headers[API_VERSION_HEADER] = SUPPORTED_VERSION
+            async with self._reported:
+                self.activity_bodies.append(json.loads(request.content))
+                self._reported.notify_all()
+            if self.hold_next_activity:
+                self.hold_next_activity = False
+                self.activity_entered.set()
+                await self.activity_release.wait()
+        return response
+
+    async def bodies_reach(self, count: int) -> list[dict[str, Any]]:
+        async with self._reported:
+            await self._reported.wait_for(lambda: len(self.activity_bodies) >= count)
+        return self.activity_bodies
+
+    async def body_at(self, index: int) -> dict[str, Any]:
+        await self.bodies_reach(index + 1)
+        return self.activity_bodies[index]
+
+    async def next_body(self) -> dict[str, Any]:
+        """Wait for the platform to receive one more activity body."""
+        return await self.body_at(len(self.activity_bodies))
+
+    async def shown(self, observation: Awaitable[None]) -> None:
+        """Make a visible change and wait until the platform has received it.
+
+        Assumes looptime: under real time every change would look late. A body
+        that arrives only after virtual time passed came from a keep-alive, not
+        the change; it is recorded with that delay in ``late_changes`` rather
+        than raised, because a handler would swallow the error as a failed turn.
+        """
+        loop = asyncio.get_running_loop()
+        index = len(self.activity_bodies)
+        observed_at = loop.time()
+        await observation
+        body = await self.body_at(index)
+        if waited := loop.time() - observed_at:
+            self.late_changes.append((body, waited))
 
     def answer(self, request: httpx.Request) -> httpx.Response:
         tail = request.url.path.rsplit("/", 1)[-1]
         if tail == "participants":
             return page([{**ALICE, "role": "member", "status": "active"}])
         if is_message_post(request):
+            self.posts.append(tail)
             self.room_messages.append(message_content(request))
             return httpx.Response(
                 201, json={"data": {"id": "m-1", "success": True, "recipients": []}}
@@ -38,6 +116,19 @@ def answer(self, request: httpx.Request) -> httpx.Response:
         return super().answer(request)
 
 
+class ObservedIdle(asyncio.Event):
+    """The idle event a graceful stop waits on, announcing that wait."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.set()
+        self.awaited = asyncio.Event()
+
+    async def wait(self) -> Literal[True]:
+        self.awaited.set()
+        return await super().wait()
+
+
 class IdleReportGate:
     """Holds the room's ``working: false`` report while ``release`` is clear."""
 
@@ -95,4 +186,4 @@ async def respond(self, request: httpx.Request) -> httpx.Response:
         if is_message_post(request) and message_content(request) == self.held_reply:
             self.reply_entered.set()
             await self.release.wait()
-        return self.answer(request)
+        return await super().respond(request)
diff --git a/tests/runtime/activityownership/test_cap.py b/tests/runtime/activityownership/test_cap.py
new file mode 100644
index 000000000..70e8b5c26
--- /dev/null
+++ b/tests/runtime/activityownership/test_cap.py
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+import pytest
+
+from band.platform.event import PlatformEvent
+from band.runtime.activity import report_tool_activity, set_permission_wait
+from band.runtime.execution import ExecutionContext
+from band.runtime.tools.agent import AgentTools
+from tests.activitysupport import IDLE_BODY, REASONING_BODY, tool_body
+from tests.conftest import make_message_event
+from tests.runtime.activityownership.kit import ActivityPlatform
+
+CAP_SECONDS = 10.0
+# Keep-alives at 3, 6 and 9 s; the one due at 12 s finds the cap passed.
+LONG_TOOL_SECONDS = 15.0
+
+
+@pytest.mark.looptime
+class TestDetailCap:
+    async def test_no_detail_is_reported_past_the_cap_until_the_turn_ends(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            await asyncio.sleep(LONG_TOOL_SECONDS)
+            await report_tool_activity(tools, "Edit")
+            await set_permission_wait(tools, True)
+            await set_permission_wait(tools, False)
+
+        ctx = execution(handler, max_working_state_seconds=CAP_SECONDS)
+        await ctx._process_event(make_message_event())
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            *[tool_body("Bash")] * 4,
+            IDLE_BODY,
+        ]
diff --git a/tests/runtime/activityownership/test_control.py b/tests/runtime/activityownership/test_control.py
new file mode 100644
index 000000000..d925bc5a4
--- /dev/null
+++ b/tests/runtime/activityownership/test_control.py
@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+import pytest
+
+from band.core.turn_lifecycle import TurnTask, bind_turn_task
+from band.platform.event import PlatformEvent
+from band.runtime.activity import report_memory_activity, report_tool_activity
+from band.runtime.execution import ExecutionContext
+from band.runtime.tools.agent import AgentTools
+from tests.activitysupport import IDLE_BODY, REASONING_BODY, tool_body
+from tests.conftest import make_message_event
+from tests.runtime.activityownership.kit import (
+    ActivityPlatform,
+    ObservedIdle,
+    past_a_keep_alive,
+)
+
+
+@pytest.mark.looptime
+class TestControlledTurnDetail:
+    async def test_a_new_message_after_an_interrupt_shows_only_its_own_detail(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        cleanup_entered = asyncio.Event()
+        cleanup_release = asyncio.Event()
+        workers: list[TurnTask] = []
+
+        async def bash(tools: AgentTools) -> None:
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                cleanup_entered.set()
+                await cleanup_release.wait()
+                await report_memory_activity(tools)
+                raise
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            if workers:
+                cleanup_release.set()
+                await asyncio.gather(workers[0].task, return_exceptions=True)
+                await activity_platform.shown(report_tool_activity(tools, "Read"))
+                return
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            workers.append(bind_turn_task(tools, asyncio.create_task(bash(tools))))
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event(msg_id="interrupted"))
+        assert ctx.interrupt()
+        await cleanup_entered.wait()
+        await ctx._process_event(make_message_event(msg_id="next"))
+        assert workers[0].task.cancelled()
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            IDLE_BODY,
+            REASONING_BODY,
+            tool_body("Read"),
+            IDLE_BODY,
+        ]
+
+    async def test_a_stop_echoed_on_a_detail_refresh_cancels_the_turn(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        captured: list[AgentTools] = []
+        cancelled = asyncio.Event()
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            captured.append(tools)
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            activity_platform.stopped = True
+            await report_tool_activity(tools, "Read")
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                await report_memory_activity(tools)
+                cancelled.set()
+                raise
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event())
+        assert cancelled.is_set()
+        await report_tool_activity(captured[0], "Edit")
+        await past_a_keep_alive()
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            tool_body("Read"),
+            IDLE_BODY,
+        ]
+
+    async def test_graceful_stop_keeps_the_draining_turn_detail_current(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        captured: list[AgentTools] = []
+        handler_started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            captured.append(AgentTools.from_context(ctx))
+            handler_started.set()
+            await release.wait()
+
+        ctx = execution(handler)
+        idle = ctx._idle_event = ObservedIdle()
+        await ctx.start()
+        await ctx.on_event(make_message_event())
+        await handler_started.wait()
+        stopping = asyncio.create_task(ctx.stop(timeout=30))
+        await idle.awaited.wait()
+        await activity_platform.shown(report_tool_activity(captured[0], "Bash"))
+        release.set()
+        assert await stopping
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            IDLE_BODY,
+        ]
+
+    async def test_an_interrupted_turn_cannot_change_detail_while_unwinding(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        cleanup_entered = asyncio.Event()
+        cleanup_release = asyncio.Event()
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                await report_memory_activity(tools)
+                cleanup_entered.set()
+                await cleanup_release.wait()
+                raise
+
+        ctx = execution(handler)
+        turn = asyncio.create_task(ctx._process_event(make_message_event()))
+        await activity_platform.bodies_reach(2)
+        ctx.interrupt()
+        await cleanup_entered.wait()
+        await past_a_keep_alive()
+        cleanup_release.set()
+        await turn
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            IDLE_BODY,
+        ]
diff --git a/tests/runtime/activityownership/test_delivery.py b/tests/runtime/activityownership/test_delivery.py
new file mode 100644
index 000000000..1e7945693
--- /dev/null
+++ b/tests/runtime/activityownership/test_delivery.py
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+import pytest
+
+from band.platform.event import PlatformEvent
+from band.runtime.activity import report_tool_activity
+from band.runtime.execution import ExecutionContext
+from band.runtime.tools.agent import AgentTools
+from tests.activitysupport import tool_body
+from tests.conftest import make_message_event
+from tests.runtime.activityownership.kit import ActivityPlatform
+
+
+@pytest.mark.looptime
+class TestNonBlockingDelivery:
+    async def test_a_tool_never_waits_for_the_platform(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        observed_before_delivery: list[bool] = []
+        seconds_until_sent: list[float] = []
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            loop = asyncio.get_running_loop()
+            activity_platform.hold_next_activity = True
+            observed_at = loop.time()
+            observing = asyncio.create_task(report_tool_activity(tools, "Bash"))
+            await activity_platform.activity_entered.wait()
+            seconds_until_sent.append(loop.time() - observed_at)
+            observed_before_delivery.append(observing.done())
+            activity_platform.activity_release.set()
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event())
+        assert observed_before_delivery == [True]
+        assert seconds_until_sent == [0]
+        assert activity_platform.activity_bodies[1] == tool_body("Bash")
diff --git a/tests/runtime/activityownership/test_detail.py b/tests/runtime/activityownership/test_detail.py
new file mode 100644
index 000000000..129ac2234
--- /dev/null
+++ b/tests/runtime/activityownership/test_detail.py
@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+import pytest
+
+from band.core.turn_lifecycle import TurnTask, bind_turn_task
+from band.platform.event import PlatformEvent
+from band.runtime.activity import (
+    end_tool_activity,
+    report_memory_activity,
+    report_tool_activity,
+    set_permission_wait,
+)
+from band.runtime.execution import ExecutionContext
+from band.runtime.tools.agent import AgentTools
+from tests.activitysupport import (
+    IDLE_BODY,
+    MEMORY_BODY,
+    PERMISSION_BODY,
+    REASONING_BODY,
+    tool_body,
+)
+from tests.conftest import make_message_event
+from tests.runtime.activityownership.kit import ActivityPlatform, past_a_keep_alive
+
+
+@pytest.mark.looptime
+class TestOwnedDetail:
+    async def test_a_turn_through_a_permission_wait_reports_each_visible_step(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            await activity_platform.shown(end_tool_activity(tools))
+            await activity_platform.shown(set_permission_wait(tools, True))
+            # Hidden behind the wait: the room keeps seeing permission.
+            await report_tool_activity(tools, "Edit")
+            await activity_platform.shown(set_permission_wait(tools, False))
+            await activity_platform.shown(report_memory_activity(tools))
+            await tools.send_message("Done", mentions=["alice"])
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event())
+        assert activity_platform.posts == ["messages"]
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            REASONING_BODY,
+            PERMISSION_BODY,
+            tool_body("Edit"),
+            MEMORY_BODY,
+            IDLE_BODY,
+        ]
+
+    async def test_a_settled_owner_keeps_reporting_its_worker_detail(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        release = asyncio.Event()
+        captured: list[AgentTools] = []
+        workers: list[TurnTask] = []
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            captured.append(tools)
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            workers.append(bind_turn_task(tools, asyncio.create_task(release.wait())))
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event())
+        owner = captured[0]
+        await activity_platform.shown(set_permission_wait(owner, True))
+        # The keep-alive repeats PERMISSION while the wait holds.
+        await activity_platform.next_body()
+        # Hidden behind the wait: the room keeps seeing permission.
+        await report_tool_activity(owner, "Edit")
+        await activity_platform.shown(set_permission_wait(owner, False))
+        release.set()
+        await workers[0].task
+        await activity_platform.next_body()
+        await ctx._join_activity_completion()
+        await end_tool_activity(owner)
+        await past_a_keep_alive()
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            PERMISSION_BODY,
+            PERMISSION_BODY,
+            tool_body("Edit"),
+            IDLE_BODY,
+        ]
+
+    async def test_a_new_turn_starts_at_reasoning(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        turns: list[ExecutionContext] = []
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            turns.append(ctx)
+            if len(turns) > 1:
+                return
+            tools = AgentTools.from_context(ctx)
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            await activity_platform.shown(set_permission_wait(tools, True))
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event(msg_id="first"))
+        await ctx._process_event(make_message_event(msg_id="second"))
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            PERMISSION_BODY,
+            IDLE_BODY,
+            REASONING_BODY,
+            IDLE_BODY,
+        ]
diff --git a/tests/runtime/activityownership/test_exclusion.py b/tests/runtime/activityownership/test_exclusion.py
new file mode 100644
index 000000000..ffcd7ccf1
--- /dev/null
+++ b/tests/runtime/activityownership/test_exclusion.py
@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import pytest
+
+from band.platform.event import PlatformEvent
+from band.runtime.activity import (
+    end_tool_activity,
+    report_memory_activity,
+    report_tool_activity,
+    set_permission_wait,
+)
+from band.runtime.execution import ExecutionContext
+from band.runtime.tools.agent import AgentTools
+from tests.conftest import make_message_event
+from tests.runtime.activityownership.kit import ActivityPlatform, past_a_keep_alive
+from tests.runtime.helpers import ROOM_ID
+
+
+@pytest.mark.looptime
+class TestExcludedReporting:
+    @pytest.mark.parametrize(
+        "settings",
+        [{"enable_working_state": False}, {"hub_room_id": ROOM_ID}],
+        ids=["disabled", "hub"],
+    )
+    async def test_excluded_reporting_ignores_observations(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+        settings: dict[str, object],
+    ) -> None:
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            await report_tool_activity(tools, "Bash")
+            await set_permission_wait(tools, True)
+            await report_memory_activity(tools)
+            await end_tool_activity(tools)
+
+        ctx = execution(handler, **settings)
+        await ctx._process_event(make_message_event())
+        await past_a_keep_alive()
+        assert not activity_platform.activity_bodies
diff --git a/tests/runtime/activityownership/test_negotiation.py b/tests/runtime/activityownership/test_negotiation.py
new file mode 100644
index 000000000..8194956a5
--- /dev/null
+++ b/tests/runtime/activityownership/test_negotiation.py
@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import httpx
+import pytest
+
+from band.platform.event import PlatformEvent
+from band.runtime.activity import report_tool_activity
+from band.runtime.execution import ExecutionContext
+from band.runtime.tools.agent import AgentTools
+from tests.activitysupport import (
+    BOOLEAN_BODY,
+    IDLE_BODY,
+    REASONING_BODY,
+    install_activity_support,
+    tool_body,
+)
+from tests.conftest import make_message_event
+from tests.runtime.activityownership.kit import ActivityPlatform
+
+
+@pytest.mark.looptime
+class TestTurnNegotiation:
+    async def test_a_fresh_process_learns_detail_from_its_first_report(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        install_activity_support(monkeypatch, learned=False)
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event())
+        assert activity_platform.activity_bodies == [
+            BOOLEAN_BODY,
+            tool_body("Bash"),
+            IDLE_BODY,
+        ]
+
+    async def test_a_rejected_detail_mid_turn_leaves_every_later_report_boolean(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            if event.payload.id == "second":
+                return
+            tools = AgentTools.from_context(ctx)
+            activity_platform.activity_responses.append(httpx.Response(422))
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            # The resend, then one keep-alive.
+            await activity_platform.bodies_reach(4)
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event(msg_id="first"))
+        await ctx._process_event(make_message_event(msg_id="second"))
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            BOOLEAN_BODY,
+            BOOLEAN_BODY,
+            IDLE_BODY,
+            BOOLEAN_BODY,
+            IDLE_BODY,
+        ]
diff --git a/tests/runtime/activityownership/test_nonowner.py b/tests/runtime/activityownership/test_nonowner.py
new file mode 100644
index 000000000..e13b7bc67
--- /dev/null
+++ b/tests/runtime/activityownership/test_nonowner.py
@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+
+import pytest
+
+from band.core.turn_lifecycle import TurnTask, bind_turn_task
+from band.platform.event import PlatformEvent
+from band.runtime.activity import report_tool_activity, set_permission_wait
+from band.runtime.execution import ExecutionContext
+from band.runtime.tools.agent import AgentTools
+from tests.activitysupport import IDLE_BODY, REASONING_BODY, tool_body
+from tests.conftest import make_message_event
+from tests.runtime.activityownership.kit import ActivityPlatform, past_a_keep_alive
+
+
+@pytest.mark.looptime
+class TestNonOwnerObservations:
+    async def test_a_busy_delivery_never_changes_the_owner_detail(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        release = asyncio.Event()
+        workers: list[TurnTask] = []
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            if workers:
+                await report_tool_activity(tools, "unrelated")
+                await set_permission_wait(tools, True)
+                return
+            await activity_platform.shown(report_tool_activity(tools, "Bash"))
+            workers.append(bind_turn_task(tools, asyncio.create_task(release.wait())))
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event(msg_id="owner"))
+        await ctx._process_event(make_message_event(msg_id="busy"))
+        await activity_platform.next_body()
+        release.set()
+        await workers[0].task
+        await activity_platform.next_body()
+        assert activity_platform.activity_bodies == [
+            REASONING_BODY,
+            tool_body("Bash"),
+            tool_body("Bash"),
+            IDLE_BODY,
+        ]
+
+    async def test_settled_turn_cannot_restart_reporting(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        captured: list[AgentTools] = []
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            captured.append(AgentTools.from_context(ctx))
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event())
+        await ctx._join_activity_completion()
+        before = list(activity_platform.activity_bodies)
+        await report_tool_activity(captured[0], "Bash")
+        await past_a_keep_alive()
+        assert before[-1] == IDLE_BODY
+        assert activity_platform.activity_bodies == before
+
+    async def test_observation_from_a_waiting_delivery_never_starts_reporting(
+        self,
+        execution: Callable[..., ExecutionContext],
+        activity_platform: ActivityPlatform,
+    ) -> None:
+        worker_release = asyncio.Event()
+        waiting_tools: list[AgentTools] = []
+        waiting_started = asyncio.Event()
+        waiting_release = asyncio.Event()
+        workers: list[TurnTask] = []
+
+        async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None:
+            tools = AgentTools.from_context(ctx)
+            if workers:
+                waiting_tools.append(tools)
+                waiting_started.set()
+                await waiting_release.wait()
+                return
+            workers.append(
+                bind_turn_task(tools, asyncio.create_task(worker_release.wait()))
+            )
+
+        ctx = execution(handler)
+        await ctx._process_event(make_message_event(msg_id="owner"))
+        waiting = asyncio.create_task(
+            ctx._process_event(make_message_event(msg_id="waiting"))
+        )
+        await waiting_started.wait()
+        worker_release.set()
+        await workers[0].task
+        await ctx._join_activity_completion()
+        before = list(activity_platform.activity_bodies)
+        await report_tool_activity(waiting_tools[0], "Bash")
+        await past_a_keep_alive()
+        assert before[-1] == IDLE_BODY
+        assert activity_platform.activity_bodies == before
+        waiting_release.set()
+        await waiting
diff --git a/tests/runtime/activityownership/test_observerless.py b/tests/runtime/activityownership/test_observerless.py
new file mode 100644
index 000000000..32e068ce5
--- /dev/null
+++ b/tests/runtime/activityownership/test_observerless.py
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import pytest
+
+from band.runtime.tools.agent import AgentTools
+from band.testing import FakeAgentTools
+from tests.runtime.activityownership.kit import (
+    ActivityPlatform,
+    past_a_keep_alive,
+    report_every_kind,
+)
+
+
+@pytest.mark.looptime
+class TestObserverlessProducers:
+    async def test_tools_outside_an_execution_report_nothing(
+        self, bare_tools: AgentTools, activity_platform: ActivityPlatform
+    ) -> None:
+        fake = FakeAgentTools()
+        await report_every_kind(fake)
+        await report_every_kind(bare_tools)
+        await past_a_keep_alive()
+        assert activity_platform.activity_bodies == []
+        assert fake.tool_calls == []
diff --git a/tests/runtime/conftest.py b/tests/runtime/conftest.py
index 46f8a06ed..f7457f7ed 100644
--- a/tests/runtime/conftest.py
+++ b/tests/runtime/conftest.py
@@ -34,6 +34,7 @@
 
 from band.core.types import PlatformConnection, PlatformMessage
 from band.runtime.presence import RoomPresence
+from tests.activitysupport import install_activity_support
 from tests.conftest import make_participant_mock
 from tests.runtime.helpers import AGENT_ID
 
@@ -259,3 +260,9 @@ def link(mock_rest_client: MagicMock) -> MagicMock:
     link.get_stale_processing_messages = AsyncMock(return_value=[])
     link.report_activity = AsyncMock(return_value=True)
     return link
+
+
+@pytest.fixture(autouse=True)
+def runtime_activity_support(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Start every runtime test on a platform that already advertised detail."""
+    install_activity_support(monkeypatch, learned=True)
diff --git a/tests/runtime/execution/test_crashrecoverysync.py b/tests/runtime/execution/test_crashrecoverysync.py
index 84bd1a499..d52e2bd01 100644
--- a/tests/runtime/execution/test_crashrecoverysync.py
+++ b/tests/runtime/execution/test_crashrecoverysync.py
@@ -2,13 +2,15 @@
 
 import asyncio
 from datetime import UTC, datetime
-from unittest.mock import AsyncMock, MagicMock
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, call
 
 import pytest
 from band_sdk_core import ClaimRegistry
 
 from band.client.streaming import MessageMetadata
 from band.platform.event import ParticipantAddedEvent, PlatformEvent
+from band.runtime import resync
 from band.runtime.execution import (
     ExecutionContext,
     ResyncOutcome,
@@ -1049,9 +1051,14 @@ async def test_refused_claims_never_spend_the_retry_budget(
         mock_handler.assert_awaited_once()
         assert not ctx._retry_tracker.is_permanently_failed(msg.id)
 
+    @pytest.mark.parametrize("initial_recovery_delay", [0.0, 10.0])
     async def test_startup_backlog_claim_failure_does_not_process_newer_ws_event(
-        self, mock_link_with_next, mock_handler
-    ):
+        self,
+        mock_link_with_next,
+        mock_handler,
+        monkeypatch: pytest.MonkeyPatch,
+        initial_recovery_delay: float,
+    ) -> None:
         """Startup sync should not switch to WebSocket after an unclaimable backlog message."""
 
         backlog_msg = PlatformMessage(
@@ -1077,6 +1084,12 @@ async def test_startup_backlog_claim_failure_does_not_process_newer_ws_event(
             ),
         )
 
+        monkeypatch.setattr(
+            resync,
+            "random",
+            SimpleNamespace(uniform=lambda lower, upper: initial_recovery_delay),
+        )
+
         await ctx.on_event(
             make_message_event(room_id="room-123", msg_id="msg-newer-ws")
         )
@@ -1085,8 +1098,10 @@ async def test_startup_backlog_claim_failure_does_not_process_newer_ws_event(
             lambda: mock_link_with_next.mark_processing.await_count >= 1
         )
 
-        mock_link_with_next.mark_processing.assert_awaited_once_with(
-            "room-123", "msg-older-claim-fails"
+        assert (
+            mock_link_with_next.mark_processing.await_args_list
+            == [call("room-123", "msg-older-claim-fails")]
+            * mock_link_with_next.mark_processing.await_count
         )
         mock_handler.assert_not_called()
         assert ctx._sync_complete is False
diff --git a/tests/runtime/workingreporter/kit.py b/tests/runtime/workingreporter/kit.py
index ee553bf4c..34d0d2b73 100644
--- a/tests/runtime/workingreporter/kit.py
+++ b/tests/runtime/workingreporter/kit.py
@@ -26,3 +26,40 @@ async def __call__(self, working: bool) -> bool:
         if self.fail:
             raise RuntimeError("report failed")
         return True
+
+
+class DetailRecorder:
+    """Report callback recording the detail a report would carry when it starts.
+
+    ``hold()`` parks the next report after recording it, until ``release()``.
+    """
+
+    def __init__(self) -> None:
+        self.detail = "reasoning"
+        self.reports: list[str | None] = []
+        self.entered = asyncio.Event()
+        self._gate: asyncio.Event | None = None
+        self._changed = asyncio.Condition()
+
+    def hold(self) -> None:
+        self._gate = asyncio.Event()
+
+    def release(self) -> None:
+        if self._gate is not None:
+            self._gate.set()
+
+    async def __call__(self, working: bool) -> bool:
+        async with self._changed:
+            self.reports.append(self.detail if working else None)
+            self._changed.notify_all()
+        gate = self._gate
+        if gate is not None:
+            self.entered.set()
+            await gate.wait()
+            self._gate = None
+        return True
+
+    async def reports_reach(self, count: int) -> list[str | None]:
+        async with self._changed:
+            await self._changed.wait_for(lambda: len(self.reports) >= count)
+        return self.reports
diff --git a/tests/runtime/workingreporter/test_refresh.py b/tests/runtime/workingreporter/test_refresh.py
new file mode 100644
index 000000000..a26606d64
--- /dev/null
+++ b/tests/runtime/workingreporter/test_refresh.py
@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from band.runtime.working_state import WorkingStateReporter
+from tests.runtime.workingreporter.kit import DetailRecorder
+
+# Longer than any test, so every report after start is a refresh.
+NO_KEEP_ALIVE = 3600.0
+
+
+class TestRefresh:
+    async def test_a_change_is_reported_without_waiting_for_the_keep_alive(
+        self,
+    ) -> None:
+        recorder = DetailRecorder()
+        reporter = WorkingStateReporter(recorder, keep_alive_seconds=NO_KEEP_ALIVE)
+        await reporter.start()
+        recorder.detail = "Bash"
+        reporter.refresh()
+        assert await recorder.reports_reach(2) == ["reasoning", "Bash"]
+        await reporter.stop()
+        assert recorder.reports == ["reasoning", "Bash", None]
+
+    async def test_a_change_during_the_first_report_follows_it(self) -> None:
+        recorder = DetailRecorder()
+        reporter = WorkingStateReporter(recorder, keep_alive_seconds=NO_KEEP_ALIVE)
+        recorder.hold()
+        starting = asyncio.create_task(reporter.start())
+        await recorder.entered.wait()
+        recorder.detail = "Bash"
+        reporter.refresh()
+        recorder.release()
+        await starting
+        assert await recorder.reports_reach(2) == ["reasoning", "Bash"]
+        await reporter.stop()
+
+    async def test_changes_during_a_report_coalesce_to_the_latest(self) -> None:
+        recorder = DetailRecorder()
+        reporter = WorkingStateReporter(recorder, keep_alive_seconds=NO_KEEP_ALIVE)
+        await reporter.start()
+        recorder.hold()
+        recorder.detail = "Read"
+        reporter.refresh()
+        await recorder.entered.wait()
+        recorder.detail = "Bash"
+        reporter.refresh()
+        recorder.detail = "memory"
+        reporter.refresh()
+        recorder.release()
+        assert await recorder.reports_reach(3) == ["reasoning", "Read", "memory"]
+        await reporter.stop()
+        assert recorder.reports == ["reasoning", "Read", "memory", None]
+
+    @pytest.mark.looptime
+    async def test_a_change_past_the_cap_is_not_reported(self) -> None:
+        recorder = DetailRecorder()
+        reporter = WorkingStateReporter(
+            recorder, keep_alive_seconds=NO_KEEP_ALIVE, max_working_state_seconds=2
+        )
+        await reporter.start()
+        await asyncio.sleep(2.1)
+        recorder.detail = "Bash"
+        reporter.refresh()
+        # The wake finds the cap passed and ends the keep-alive without a report.
+        await reporter._task
+        await reporter.stop()
+        assert recorder.reports == ["reasoning", None]
diff --git a/tests/runtime/workingreporter/test_startstop.py b/tests/runtime/workingreporter/test_startstop.py
index 3d87f3424..86dbc080c 100644
--- a/tests/runtime/workingreporter/test_startstop.py
+++ b/tests/runtime/workingreporter/test_startstop.py
@@ -5,7 +5,7 @@
 import pytest
 
 from band.runtime.working_state import WorkingStateReporter
-from tests.runtime.workingreporter.kit import Recorder
+from tests.runtime.workingreporter.kit import DetailRecorder, Recorder
 
 
 class TestStartStop:
@@ -50,3 +50,18 @@ async def test_disabled_reports_nothing(self):
         await reporter.stop()
 
         assert rec.calls == []
+
+    @pytest.mark.looptime
+    async def test_stop_during_the_first_report_leaves_no_keep_alive(self) -> None:
+        recorder = DetailRecorder()
+        reporter = WorkingStateReporter(recorder, keep_alive_seconds=3.0)
+        recorder.hold()
+        starting = asyncio.create_task(reporter.start())
+        await recorder.entered.wait()
+        stopping = asyncio.create_task(reporter.stop())
+        recorder.release()
+        await asyncio.gather(starting, stopping)
+        await reporter.start()
+        await asyncio.sleep(3.1)
+        await reporter.stop()
+        assert recorder.reports == ["reasoning", None, "reasoning", "reasoning", None]