fix: keep activity reporting alive for the provider turn

docs/turn-outcome.md

@@ -374,7 +374,7 @@ refusal instead of returning `DONE`; a drain records only accepted messages.
 
 ## Cancellation ownership
 
-Each runtime handler and SDK-managed Codex/Claude worker has one task owner.
+Each runtime handler and SDK-managed Claude, Codex or OpenCode worker has one task owner.
 The originating turn scope coordinates termination and fixes one absolute cleanup
 deadline using the existing one-second grace. Repeated controls still notify the
 adapter and update the unsettled delivery outcome, but never cancel its cleanup
@@ -384,9 +384,9 @@ A stopped tool or auxiliary claim first revokes that scope's posting permission
 and raises cancellation in its caller. Bound workers and the handler get the same
 cleanup grace to unwind cooperatively; an unfinished task receives cancellation
 once at expiry if none was delivered, even when nothing waits for it. Native
-controls, heartbeat refusal, watchdog expiry and shutdown cancel a running turn
-immediately; a worker already released for approval is left to the adapter's
-room cleanup. Provider execution and recovery timeouts retain their adapter
+STOP and INTERRUPT controls, heartbeat refusal and watchdog expiry cancel the
+originating turn, including a worker released for approval. Leave and shutdown
+clear reporting while released worker cleanup remains with adapter residency. Provider execution and recovery timeouts retain their adapter
 policy and coordinate with that same owner.
 
 Unfinished tasks remain owned after grace expires. They cannot post late replies,
@@ -410,3 +410,17 @@ and never awaited on the control-processing task. A timeout branch awaits
 `stop` before its own cleanup and report. The room's next
 turn waits for `settled(room_id)`, and teardown for `settled` or `drain()`,
 before reacquiring or closing the backend.
+
+## Working-state lifetime
+
+The activity owner is the originating turn scope, independently of the current
+message delivery. A registered provider worker keeps its reporter alive after
+the handler settles, including while waiting for human approval. Later approval
+replies and busy deliveries do not replace that owner or restart its duration
+cap. Reporting ends when both the handler and its workers finish.
+
+An echoed heartbeat STOP can cancel that detached provider and notify its adapter
+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.

docs/websocket-events.md

@@ -49,7 +49,10 @@ While working-state reporting is active, its existing activity keep-alive also
 recovers a missed STOP: the platform answers a `working: true` report with
 `working: false` when the room execution is stopped. The runtime cancels the
 originating turn through the same platform-stop path as a refused claim or post,
-leaving its message PROCESSING for later replay. Responses from before PLAY,
+leaving an unsettled message PROCESSING for later replay. A provider worker
+released for approval retains its originating activity scope across later
+deliveries; a late heartbeat STOP cancels that provider without changing the
+original message's settled acknowledgement. Responses from before PLAY,
 a replacement turn, or shutdown cannot stop newer work. Request failures never
 count as STOP, and the public activity-report boolean still means request success.
 
@@ -94,7 +97,9 @@ never interprets approval syntax or drains model context for a control.
 
 Room admission and control revisions belong to the execution context; a turn
 scope owns termination intent and delivery disposition. Each concrete task has
-one cancellation owner, including explicitly bound Codex/Claude workers.
+one cancellation owner, including explicitly bound Claude, Codex and OpenCode
+workers. Activity follows that scope until its handler and registered workers
+finish, independently of current delivery admission.
 Shutdown takes precedence; otherwise the latest accepted native STOP/INTERRUPT
 before settlement determines delivery. An observed REST refusal cannot replace
 an accepted native disposition. PLAY advances admission without reviving an

src/band/adapters/opencode/adapter.py

@@ -28,7 +28,7 @@
 )
 from band.core.simple_adapter import SimpleAdapter
 from band.core.turn import judge_detached_turn
-from band.core.turn_lifecycle import BackendTurns
+from band.core.turn_lifecycle import BackendTurns, bind_turn_task
 from band.core.types import (
     AdapterFeatures,
     Capability,
@@ -495,6 +495,7 @@ async def on_message(
                 self._watch_turn_completion(room_state, room_id, turn)
             )
             turn.watcher = turn_task
+            bind_turn_task(tools, turn_task)
             try:
                 # Turn-phase diagnostics (classify a stuck turn from CI logs):
                 # if 'returned' never follows 'start', prompt_async is blocking

src/band/runtime/execution.py

@@ -414,6 +414,9 @@ def __init__(
 
         self._detached_cycle_tasks: dict[asyncio.Task[None], TurnTask] = {}
         self.current_scope: TurnScope | None = None
+        self._activity_owner: TurnScope | None = None
+        self._activity_stop_scope: TurnScope | None = None
+        self._activity_completion: asyncio.Task[None] | None = None
         self._control_revision = 0
         self._on_platform_stop = on_platform_stop
         self._shutdown_task: asyncio.Task[bool] | None = None
@@ -452,8 +455,10 @@ def is_stopped(self) -> bool:
     @property
     def adapter_stop_notified(self) -> bool:
         """Whether an observed STOP already notified the adapter for this turn."""
-        scope = self.current_scope
-        return scope is not None and scope.observer_task is not None
+        return any(
+            scope is not None and scope.observer_task is not None
+            for scope in (self.current_scope, self._activity_stop_scope)
+        )
 
     def _set_state(self, new_state: ExecutionState) -> None:
         """
@@ -630,6 +635,8 @@ async def _stop(self, timeout: float | None) -> bool:
             loop_owner.terminate()
             await self._drain_or_retain(loop_owner)
             self._process_loop_task = None
+        self._activity_owner = None
+        await self._join_activity_completion()
         await self._stop_retained_tasks(terminal_deadline())
         await self._working_reporter.stop()
         return self._stop_graceful
@@ -737,10 +744,13 @@ def _active_cycle_task(self) -> asyncio.Task[None] | None:
 
     def _cancel_cycle(self, kind: ControlMode) -> bool:
         scope = self._running_scope
-        if scope is None:
-            return False
-        scope.terminate(kind)
-        return True
+        activity = self._activity_owner
+        if activity is not None and activity is not scope:
+            activity.terminate(kind)
+            self._finish_activity(activity, aborted=True)
+        if scope is not None:
+            scope.terminate(kind)
+        return scope is not None or activity is not None
 
     def stop_room(self) -> None:
         """Pause triggers and abort the cycle, leaving it actionable for PLAY."""
@@ -768,7 +778,11 @@ def _reset_resync_interval(self) -> None:
             self._queue_wait.reschedule(self._resync_schedule.due_at)
 
     def owns_stop_observation(self, scope: TurnScope) -> bool:
-        return self._owns_control_revision(scope, scope.stop_revision)
+        return (
+            scope
+            in (self.current_scope, self._activity_owner, self._activity_stop_scope)
+            and scope.stop_revision == self._control_revision
+        )
 
     def _owns_control_revision(self, scope: TurnScope, revision: int | None) -> bool:
         return self.current_scope is scope and revision == self._control_revision
@@ -785,8 +799,10 @@ def observe_platform_stop(
                 scope.control_revision if control_revision is None else control_revision
             )
         scope.terminate(ControlMode.STOP, cooperative=cooperative, observed=True)
-        if self.current_scope is not scope:
-            return
+        if scope is self._activity_owner:
+            self._activity_stop_scope = scope
+            if scope.settled:
+                self._finish_activity(scope, aborted=True)
         if (
             not self._closing
             and self.owns_stop_observation(scope)
@@ -1864,14 +1880,19 @@ async def _report_working_state(self, working: bool) -> bool:
             working,
             timeout_seconds=self.config.working_request_timeout_seconds,
             on_stopped=partial(
-                self._observe_activity_stop, self.current_scope, self._control_revision
+                self._observe_activity_stop,
+                self._activity_owner or self.current_scope,
+                self._control_revision,
             ),
         )
 
     def _observe_activity_stop(self, scope: TurnScope | None, revision: int) -> None:
         if (
             scope is None
-            or not self._owns_control_revision(scope, revision)
+            or not (
+                scope in (self._activity_owner, self.current_scope)
+                and revision == self._control_revision
+            )
             or scope.stop_observed
             or self._closing
             or scope.kind is not None
@@ -1885,15 +1906,83 @@ def _observe_activity_stop(self, scope: TurnScope | None, revision: int) -> None
     async def _execute_message_cycle(
         self, event: PlatformEvent, message_id: str | None
     ) -> None:
-        await self._working_reporter.start()
         scope = self.current_scope
+        if scope is not None:
+            await self._acquire_activity(scope)
         if scope is not None and scope.terminating:
             raise asyncio.CancelledError
         self._resync_handled_work = True
         self._reset_resync_interval()
         with _observe_phase("handler", self.room_id, message_id):
             await self._on_execute(self, event)
 
+    async def _join_activity_completion(self) -> None:
+        task = self._activity_completion
+        if task is not None:
+            await asyncio.shield(task)
+            if self._activity_completion is task:
+                self._activity_completion = None
+
+    async def _acquire_activity(self, scope: TurnScope) -> None:
+        await self._join_activity_completion()
+        await self._start_activity(scope)
+
+    async def _start_activity(self, scope: TurnScope) -> None:
+        previous = self._activity_owner
+        if previous is not None and previous is not scope:
+            if not previous.aborted and any(
+                not worker.task.done() for worker in previous.workers
+            ):
+                return
+            self._finish_activity(previous, aborted=True)
+            await self._join_activity_completion()
+        if scope.aborted or self._closing:
+            return
+        if self._activity_owner is not scope:
+            self._activity_stop_scope = None
+        self._activity_owner = scope
+        await self._working_reporter.start()
+
+    def register_turn_task(self, scope: TurnScope, owner: TurnTask) -> None:
+        """Bind completion to the same owner that delivers worker cancellation."""
+        scope.bind(owner)
+        owner.task.add_done_callback(lambda _: self._finish_activity(scope))
+        if scope is self.current_scope and self._activity_owner is None:
+            previous = self._activity_completion
+
+            async def acquire() -> None:
+                if previous is not None:
+                    await asyncio.shield(previous)
+                if scope is self.current_scope:
+                    await self._start_activity(scope)
+
+            self._retain_activity_completion(asyncio.create_task(acquire()))
+
+    def _retain_activity_completion(self, task: asyncio.Task[None]) -> None:
+        self._activity_completion = task
+        task.add_done_callback(self._activity_completed)
+
+    def _activity_completed(self, task: asyncio.Task[None]) -> None:
+        task_failure(task)
+        if self._activity_completion is task:
+            self._activity_completion = None
+            if self._activity_owner is not None:
+                self._finish_activity(self._activity_owner)
+
+    def _finish_activity(self, scope: TurnScope, *, aborted: bool = False) -> None:
+        if self._activity_owner is not scope:
+            return
+        if not aborted and (
+            not scope.settled or any(not owner.task.done() for owner in scope.workers)
+        ):
+            return
+        self._activity_owner = None
+        self._working_reporter.defer_false_report()
+        if self._activity_completion is None:
+            self._retain_activity_completion(
+                asyncio.create_task(self._working_reporter.stop())
+            )
+
     def _retain_detached_cycle_task(
         self, task: asyncio.Task[None], *, owner: TurnTask | None = None
     ) -> None:
@@ -1979,15 +2068,11 @@ async def invoke() -> None:
         finally:
             wake.cancel()
             await asyncio.gather(wake, return_exceptions=True)
-            if isinstance(event, MessageEvent) and handler is None:
-                # stop() joins this cycle inside the settlement budget, which is
-                # shorter than the activity POST. A closing execution finishes
-                # working:false after that join.
-                if self._closing:
-                    self._working_reporter.defer_false_report()
-                else:
-                    await self._working_reporter.stop()
             scope.settled = True
+            if isinstance(event, MessageEvent) and handler is None:
+                self._finish_activity(scope, aborted=scope.terminating)
+                if not self._closing:
+                    await self._join_activity_completion()
         if scope.shutdown:
             raise asyncio.CancelledError
         if scope.kind is not None:

src/band/runtime/tools/agent.py

@@ -389,7 +389,10 @@ def from_context(
 
     def register_turn_task(self, owner: TurnTask) -> None:
         if self._scope is not None:
-            self._scope.bind(owner)
+            if self._ctx is None:
+                self._scope.bind(owner)
+            else:
+                self._ctx.register_turn_task(self._scope, owner)
 
     def _abort_stopped_post(self, error: RoomExecutionStoppedError) -> None:
         if self._ctx is not None and self._scope is not None:

src/band/runtime/working_state.py

@@ -74,6 +74,7 @@ async def start(self) -> None:
             # Re-entrant start while active is a no-op: the signal is already
             # alive and the keep-alive task is already running.
             return
+        await self._finish_false_send()
         self._active = True
         self._started_at = asyncio.get_running_loop().time()
         await self._safe_report(True)

tests/runtime/activityownership/__init__.py

@@ -0,0 +1 @@
+from __future__ import annotations

tests/runtime/activityownership/conftest.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Awaitable, Callable
+from unittest.mock import AsyncMock
+
+import pytest
+
+from band.platform.event import PlatformEvent
+from band.platform.link import BandLink
+from band.runtime.execution import ExecutionContext
+from band.runtime.runtime import AgentRuntime
+from band.runtime.sessionconfig import SessionConfig
+from tests.runtime.activityownership.kit import ActivityPlatform
+from tests.runtime.helpers import (
+    AGENT_ID,
+    ROOM_ID,
+    LifecyclePlatform,
+    rest_client_over,
+)
+
+
+@pytest.fixture
+async def activity_platform() -> AsyncIterator[LifecyclePlatform]:
+    yield ActivityPlatform()
+
+
+@pytest.fixture
+async def execution(
+    activity_platform: LifecyclePlatform,
+) -> AsyncIterator[Callable[..., ExecutionContext]]:
+    contexts: list[ExecutionContext] = []
+    async with rest_client_over(activity_platform.answer) as rest:
+
+        def create(
+            handler: Callable[[ExecutionContext, PlatformEvent], Awaitable[None]],
+            **settings: object,
+        ) -> ExecutionContext:
+            link = BandLink(agent_id=AGENT_ID, api_key="test-key")
+            link.rest = rest
+            runtime = AgentRuntime(
+                link=link, agent_id=AGENT_ID, on_execute=handler, on_control=AsyncMock()
+            )
+            ctx = ExecutionContext(
+                ROOM_ID,
+                link,
+                handler,
+                agent_id=AGENT_ID,
+                on_platform_stop=runtime._on_platform_stop,
+                config=SessionConfig(enable_context_hydration=False, **settings),
+            )
+            contexts.append(ctx)
+            return ctx
+
+        yield create
+        for ctx in contexts:
+            await ctx.stop()

tests/runtime/activityownership/kit.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+import httpx
+
+from tests.runtime.helpers import ALICE, LifecyclePlatform, page
+
+
+class ActivityPlatform(LifecyclePlatform):
+    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 request.method == "POST" and tail == "messages":
+            return httpx.Response(
+                201, json={"data": {"id": "m-1", "success": True, "recipients": []}}
+            )
+        return super().answer(request)