diff --git a/docs/turn-outcome.md b/docs/turn-outcome.md index 54e637d36..6324fa535 100644 --- a/docs/turn-outcome.md +++ b/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. diff --git a/docs/websocket-events.md b/docs/websocket-events.md index f233da3fb..f623f8cd4 100644 --- a/docs/websocket-events.md +++ b/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 diff --git a/src/band/adapters/opencode/adapter.py b/src/band/adapters/opencode/adapter.py index 7d510a6dc..98fbb2136 100644 --- a/src/band/adapters/opencode/adapter.py +++ b/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 diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index 701180cca..971ee0d0f 100644 --- a/src/band/runtime/execution.py +++ b/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,8 +1906,9 @@ 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 @@ -1894,6 +1916,73 @@ async def _execute_message_cycle( 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: diff --git a/src/band/runtime/tools/agent.py b/src/band/runtime/tools/agent.py index deeaa144e..88d53634b 100644 --- a/src/band/runtime/tools/agent.py +++ b/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: diff --git a/src/band/runtime/working_state.py b/src/band/runtime/working_state.py index f9402c389..4d842c7fe 100644 --- a/src/band/runtime/working_state.py +++ b/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) diff --git a/tests/runtime/activityownership/__init__.py b/tests/runtime/activityownership/__init__.py new file mode 100644 index 000000000..9d48db4f9 --- /dev/null +++ b/tests/runtime/activityownership/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/tests/runtime/activityownership/conftest.py b/tests/runtime/activityownership/conftest.py new file mode 100644 index 000000000..16bcf6bfc --- /dev/null +++ b/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() diff --git a/tests/runtime/activityownership/kit.py b/tests/runtime/activityownership/kit.py new file mode 100644 index 000000000..d43a5b921 --- /dev/null +++ b/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) diff --git a/tests/runtime/activityownership/test_lifetime.py b/tests/runtime/activityownership/test_lifetime.py new file mode 100644 index 000000000..225251892 --- /dev/null +++ b/tests/runtime/activityownership/test_lifetime.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Callable + +import pytest + +from band.client.streaming import ControlMode +from band.core.turn_lifecycle import TurnTask, bind_turn_task +from band.platform.event import PlatformEvent +from band.runtime.execution import ExecutionContext +from band.runtime.tools.agent import AgentTools +from tests.conftest import make_message_event +from tests.runtime.helpers import IDLE, KEEP_ALIVE, LifecyclePlatform + + +class TestProviderLifetime: + @pytest.mark.looptime + async def test_detached_worker_keeps_heartbeat_across_busy_delivery( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + ) -> None: + release = asyncio.Event() + workers: list[TurnTask] = [] + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + if workers: + return + tools = AgentTools.from_context(ctx) + workers.append(bind_turn_task(tools, asyncio.create_task(release.wait()))) + + ctx = execution(handler) + await ctx._process_event(make_message_event(msg_id="original")) + await asyncio.sleep(3.1) + await ctx._process_event(make_message_event(msg_id="busy")) + assert activity_platform.activity_reports == [KEEP_ALIVE, KEEP_ALIVE] + assert activity_platform.marked("processed") == ["original", "busy"] + release.set() + await workers[0].task + await asyncio.sleep(0) + await ctx._join_activity_completion() + assert activity_platform.activity_reports[-1] == IDLE + + async def test_worker_completion_does_not_clear_running_handler( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + ) -> None: + finished, release = asyncio.Event(), asyncio.Event() + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + worker = asyncio.create_task(asyncio.sleep(0)) + bind_turn_task(AgentTools.from_context(ctx), worker) + await worker + finished.set() + await release.wait() + + ctx = execution(handler) + turn = asyncio.create_task(ctx._process_event(make_message_event())) + await finished.wait() + assert activity_platform.activity_reports == [KEEP_ALIVE] + release.set() + await turn + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] + + @pytest.mark.parametrize("mode", [ControlMode.STOP, ControlMode.INTERRUPT]) + async def test_detached_control_cancels_worker_preserving_settled_ack( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + mode: ControlMode, + ) -> None: + workers: list[TurnTask] = [] + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + workers.append( + bind_turn_task( + AgentTools.from_context(ctx), + asyncio.create_task(asyncio.Event().wait()), + ) + ) + + ctx = execution(handler) + await ctx._process_event(make_message_event(msg_id="original")) + assert ctx.interrupt(kind=mode) + await asyncio.gather(workers[0].task, return_exceptions=True) + await ctx._join_activity_completion() + assert workers[0].task.cancelled() + assert activity_platform.marked("processed") == ["original"] + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] + + @pytest.mark.looptime + async def test_detached_heartbeat_stop_preserves_ack( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + ) -> None: + workers: list[TurnTask] = [] + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + if not workers: + workers.append( + bind_turn_task( + AgentTools.from_context(ctx), + asyncio.create_task(asyncio.Event().wait()), + ) + ) + + ctx = execution(handler) + await ctx._process_event(make_message_event(msg_id="original")) + await ctx._process_event(make_message_event(msg_id="busy")) + activity_platform.stopped = True + await asyncio.sleep(3.1) + assert workers[0].task.cancelled() + assert activity_platform.marked("processed") == ["original", "busy"] + assert activity_platform.activity_reports[-1] == IDLE + + @pytest.mark.looptime + async def test_busy_delivery_keeps_original_cap_deadline( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + ) -> None: + release = asyncio.Event() + workers: list[TurnTask] = [] + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + if not workers: + workers.append( + bind_turn_task( + AgentTools.from_context(ctx), + asyncio.create_task(release.wait()), + ) + ) + + ctx = execution(handler, max_working_state_seconds=4.0) + await ctx._process_event(make_message_event(msg_id="original")) + await asyncio.sleep(3.1) + await ctx._process_event(make_message_event(msg_id="busy")) + await asyncio.sleep(6) + assert activity_platform.activity_reports == [KEEP_ALIVE, KEEP_ALIVE] + assert not workers[0].task.done() + release.set() + await workers[0].task + await asyncio.sleep(0) + await ctx._join_activity_completion() + assert activity_platform.activity_reports[-1] == IDLE + + async def test_shutdown_clears_reporting_without_cancelling_detached_worker( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + ) -> None: + release = asyncio.Event() + workers: list[TurnTask] = [] + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + workers.append( + bind_turn_task( + AgentTools.from_context(ctx), asyncio.create_task(release.wait()) + ) + ) + + ctx = execution(handler) + await ctx._process_event(make_message_event()) + await ctx.stop() + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] + assert not workers[0].task.done() + release.set() + await workers[0].task + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] + + async def test_new_worker_admitted_after_previous_completion_gets_own_indicator( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + ) -> None: + releases = [asyncio.Event(), asyncio.Event()] + entered, admitted = asyncio.Event(), asyncio.Event() + workers: list[TurnTask] = [] + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + index = len(workers) + if index: + entered.set() + await admitted.wait() + workers.append( + bind_turn_task( + AgentTools.from_context(ctx), + asyncio.create_task(releases[index].wait()), + ) + ) + + ctx = execution(handler) + await ctx._process_event(make_message_event(msg_id="original")) + turn = asyncio.create_task(ctx._process_event(make_message_event(msg_id="new"))) + await entered.wait() + releases[0].set() + await workers[0].task + await asyncio.sleep(0) + await ctx._join_activity_completion() + admitted.set() + await turn + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE, KEEP_ALIVE] + releases[1].set() + await workers[1].task + await asyncio.sleep(0) + await ctx._join_activity_completion() + assert activity_platform.activity_reports == [ + KEEP_ALIVE, + IDLE, + KEEP_ALIVE, + IDLE, + ] diff --git a/tests/runtime/activityownership/test_opencode.py b/tests/runtime/activityownership/test_opencode.py new file mode 100644 index 000000000..e96a45859 --- /dev/null +++ b/tests/runtime/activityownership/test_opencode.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from unittest.mock import patch + +import pytest + +from band.adapters.opencode import OpencodeAdapter +from band.client.streaming import ControlMode +from band.integrations.opencode.types import OpencodeSessionState +from band.platform.event import PlatformEvent +from band.runtime.execution import ExecutionContext +from band.runtime.tools.agent import AgentTools +from tests.adapters.opencode.helpers import ( + FakeOpencodeClient, + event_permission, + make_platform_message, +) +from tests.conftest import make_message_event +from tests.mcpbackends import backends_created_by +from tests.runtime.helpers import IDLE, KEEP_ALIVE, ROOM_ID, LifecyclePlatform + + +class TestOpenCodeLifetime: + @pytest.mark.parametrize("mode", [ControlMode.STOP, ControlMode.INTERRUPT]) + async def test_parked_native_turn_dual_dispatch_aborts_remote_once( + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, + mode: ControlMode, + ) -> None: + client = FakeOpencodeClient( + prompt_event_sequences=[[event_permission("sess-1", "permission-1")]] + ) + adapter = OpencodeAdapter(client_factory=lambda _: client, emit=()) + aborting, release, cleaned = (asyncio.Event() for _ in range(3)) + original_abort = client.abort_session + + async def abort(session_id: str) -> None: + aborting.set() + await release.wait() + await original_abort(session_id) + cleaned.set() + + async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: + await adapter.on_message( + make_platform_message(room_id=ROOM_ID, sender_id="user-456"), + AgentTools.from_context(ctx), + OpencodeSessionState(), + None, + None, + is_session_bootstrap=True, + room_id=ROOM_ID, + ) + + with backends_created_by(), patch.object(client, "abort_session", abort): + await adapter.on_started("Test", "Test") + ctx = execution(handler) + try: + await ctx._process_event(make_message_event(msg_id="original")) + assert activity_platform.activity_reports == [KEEP_ALIVE] + assert activity_platform.marked("processed") == ["original"] + assert ctx.interrupt(kind=mode) + notify = asyncio.create_task(adapter.on_interrupt(ROOM_ID, mode)) + await aborting.wait() + assert not cleaned.is_set() + release.set() + await notify + await adapter.backend_turns.settled(ROOM_ID) + await ctx._join_activity_completion() + assert client.aborted_sessions == ["sess-1"] + assert cleaned.is_set() + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] + assert activity_platform.marked("processed") == ["original"] + finally: + release.set() + await adapter.on_cleanup(ROOM_ID) diff --git a/tests/runtime/helpers.py b/tests/runtime/helpers.py index e2ad63b61..590062e33 100644 --- a/tests/runtime/helpers.py +++ b/tests/runtime/helpers.py @@ -299,13 +299,16 @@ class PostingPlatform(LifecyclePlatform): def answer(self, request: httpx.Request) -> httpx.Response: tail = request.url.path.rsplit("/", 1)[-1] if tail == "participants": - return page([ALICE]) + return page([{**ALICE, "role": "member", "status": "active"}]) if request.method == "POST" and tail == "events": return httpx.Response( - 201, json={"data": {"id": "e-1", "message_type": "event"}} + 201, + json={"data": {"id": "e-1", "message_type": "event", "success": True}}, ) if request.method == "POST" and tail == "messages": - return httpx.Response(201, json={"data": {"id": "m-1", "recipients": []}}) + return httpx.Response( + 201, json={"data": {"id": "m-1", "recipients": [], "success": True}} + ) return super().answer(request) diff --git a/tests/runtime/test_working_state.py b/tests/runtime/test_working_state.py deleted file mode 100644 index 8b0a1bd02..000000000 --- a/tests/runtime/test_working_state.py +++ /dev/null @@ -1,203 +0,0 @@ -"""Tests for WorkingStateReporter (boolean working-state keep-alive).""" - -from __future__ import annotations - -import asyncio - -import pytest - -from band.runtime.working_state import WorkingStateReporter - - -async def _wait_for(predicate, timeout: float = 2.0) -> None: - """Poll until predicate() is truthy or raise on timeout (non-brittle wait).""" - loop = asyncio.get_running_loop() - deadline = loop.time() + timeout - while loop.time() < deadline: - if predicate(): - return - await asyncio.sleep(0.005) - raise AssertionError("condition not met within timeout") - - -class Recorder: - """Async report callback that records the sequence of working values.""" - - def __init__(self, *, fail: bool = False) -> None: - self.calls: list[bool] = [] - self.fail = fail - - async def __call__(self, working: bool) -> bool: - self.calls.append(working) - if self.fail: - raise RuntimeError("report failed") - return True - - -class TestStartStop: - @pytest.mark.asyncio - async def test_start_sends_true_immediately(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) - - await reporter.start() - - assert rec.calls == [True] - await reporter.stop() - - @pytest.mark.asyncio - async def test_stop_sends_false(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) - - await reporter.start() - await reporter.stop() - - assert rec.calls[0] is True - assert rec.calls[-1] is False - assert rec.calls.count(False) == 1 - - @pytest.mark.asyncio - async def test_stop_without_start_is_noop(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) - - await reporter.stop() - - assert rec.calls == [] - - @pytest.mark.asyncio - async def test_disabled_reports_nothing(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=0.01, enabled=False) - - await reporter.start() - await asyncio.sleep(0.05) - await reporter.stop() - - assert rec.calls == [] - - -class TestKeepAlive: - @pytest.mark.asyncio - async def test_refreshes_true_on_cadence(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=0.02) - - await reporter.start() - # initial true + at least two refreshes - await _wait_for(lambda: rec.calls.count(True) >= 3) - await reporter.stop() - - assert rec.calls.count(True) >= 3 - assert rec.calls[-1] is False - - @pytest.mark.asyncio - async def test_reentrant_start_does_not_duplicate_keepalive(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) - - await reporter.start() - await reporter.start() # already active -> no-op - - assert rec.calls == [True] # only one initial true, one task - await reporter.stop() - - -class TestOrdering: - @pytest.mark.asyncio - async def test_false_always_after_true_even_on_ultra_short_cycle(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) - - # Ultra-short cycle: start then immediately stop. - await reporter.start() - await reporter.stop() - - # The lock must guarantee true is sent (and lands) before false. - assert rec.calls[0] is True - assert rec.calls[-1] is False - # No false appears before the first true. - first_false = rec.calls.index(False) - first_true = rec.calls.index(True) - assert first_true < first_false - - -class TestCancellation: - @pytest.mark.asyncio - async def test_false_sent_even_when_cycle_cancelled(self): - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) - - async def cycle(): - await reporter.start() - try: - await asyncio.sleep(100) # long-running reasoning - finally: - await reporter.stop() - - task = asyncio.create_task(cycle()) - await _wait_for(lambda: rec.calls == [True]) - - task.cancel() - with pytest.raises(asyncio.CancelledError): - await task - - # The shielded false-send must have completed despite the cancel. - await _wait_for(lambda: rec.calls and rec.calls[-1] is False) - assert rec.calls.count(False) == 1 - - def test_stop_sends_false_when_the_keepalive_loop_has_closed(self) -> None: - rec = Recorder() - reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) - with asyncio.Runner() as runner: - runner.run(reporter.start()) - - resumed = asyncio.new_event_loop() - try: - resumed.run_until_complete(reporter.stop()) - finally: - resumed.close() - - assert rec.calls == [True, False] - - -class TestFailureSwallowed: - @pytest.mark.asyncio - async def test_report_exception_does_not_propagate(self): - rec = Recorder(fail=True) - reporter = WorkingStateReporter(rec, keep_alive_seconds=0.02) - - # Neither start nor the keep-alive task nor stop may raise. - await reporter.start() - await asyncio.sleep(0.05) - await reporter.stop() - - assert rec.calls # attempts were made - - -class TestMaxDuration: - @pytest.mark.asyncio - async def test_keepalive_stops_after_max_but_stop_still_sends_false(self): - rec = Recorder() - reporter = WorkingStateReporter( - rec, keep_alive_seconds=0.02, max_working_state_seconds=0.05 - ) - - await reporter.start() - # Let the cap elapse and the keep-alive go quiet. - await asyncio.sleep(0.2) - count_after_cap = rec.calls.count(True) - await asyncio.sleep(0.1) - # No further true pings after the cap. - assert rec.calls.count(True) == count_after_cap - - # stop() still sends the authoritative false. - await reporter.stop() - assert rec.calls[-1] is False - - -class TestValidation: - def test_non_positive_cadence_rejected(self): - with pytest.raises(ValueError): - WorkingStateReporter(Recorder(), keep_alive_seconds=0) diff --git a/tests/runtime/workingreporter/__init__.py b/tests/runtime/workingreporter/__init__.py new file mode 100644 index 000000000..9d48db4f9 --- /dev/null +++ b/tests/runtime/workingreporter/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/tests/runtime/workingreporter/kit.py b/tests/runtime/workingreporter/kit.py new file mode 100644 index 000000000..ee553bf4c --- /dev/null +++ b/tests/runtime/workingreporter/kit.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +import asyncio + + +async def _wait_for(predicate, timeout: float = 2.0) -> None: + """Poll until predicate() is truthy or raise on timeout (non-brittle wait).""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if predicate(): + return + await asyncio.sleep(0.005) + raise AssertionError("condition not met within timeout") + + +class Recorder: + """Async report callback that records the sequence of working values.""" + + def __init__(self, *, fail: bool = False) -> None: + self.calls: list[bool] = [] + self.fail = fail + + async def __call__(self, working: bool) -> bool: + self.calls.append(working) + if self.fail: + raise RuntimeError("report failed") + return True diff --git a/tests/runtime/workingreporter/test_cancellation.py b/tests/runtime/workingreporter/test_cancellation.py new file mode 100644 index 000000000..a056949fc --- /dev/null +++ b/tests/runtime/workingreporter/test_cancellation.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingreporter.kit import Recorder, _wait_for + + +class TestCancellation: + @pytest.mark.asyncio + async def test_false_sent_even_when_cycle_cancelled(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) + + async def cycle(): + await reporter.start() + try: + await asyncio.sleep(100) # long-running reasoning + finally: + await reporter.stop() + + task = asyncio.create_task(cycle()) + await _wait_for(lambda: rec.calls == [True]) + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # The shielded false-send must have completed despite the cancel. + await _wait_for(lambda: rec.calls and rec.calls[-1] is False) + assert rec.calls.count(False) == 1 + + def test_stop_sends_false_when_the_keepalive_loop_has_closed(self) -> None: + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) + with asyncio.Runner() as runner: + runner.run(reporter.start()) + + resumed = asyncio.new_event_loop() + try: + resumed.run_until_complete(reporter.stop()) + finally: + resumed.close() + + assert rec.calls == [True, False] diff --git a/tests/runtime/workingreporter/test_failureswallowed.py b/tests/runtime/workingreporter/test_failureswallowed.py new file mode 100644 index 000000000..099a8c6a6 --- /dev/null +++ b/tests/runtime/workingreporter/test_failureswallowed.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingreporter.kit import Recorder + + +class TestFailureSwallowed: + @pytest.mark.asyncio + async def test_report_exception_does_not_propagate(self): + rec = Recorder(fail=True) + reporter = WorkingStateReporter(rec, keep_alive_seconds=0.02) + + # Neither start nor the keep-alive task nor stop may raise. + await reporter.start() + await asyncio.sleep(0.05) + await reporter.stop() + + assert rec.calls # attempts were made diff --git a/tests/runtime/workingreporter/test_keepalive.py b/tests/runtime/workingreporter/test_keepalive.py new file mode 100644 index 000000000..5de332799 --- /dev/null +++ b/tests/runtime/workingreporter/test_keepalive.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingreporter.kit import Recorder, _wait_for + + +class TestKeepAlive: + @pytest.mark.asyncio + async def test_refreshes_true_on_cadence(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=0.02) + + await reporter.start() + # initial true + at least two refreshes + await _wait_for(lambda: rec.calls.count(True) >= 3) + await reporter.stop() + + assert rec.calls.count(True) >= 3 + assert rec.calls[-1] is False + + @pytest.mark.asyncio + async def test_reentrant_start_does_not_duplicate_keepalive(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) + + await reporter.start() + await reporter.start() # already active -> no-op + + assert rec.calls == [True] # only one initial true, one task + await reporter.stop() diff --git a/tests/runtime/workingreporter/test_maxduration.py b/tests/runtime/workingreporter/test_maxduration.py new file mode 100644 index 000000000..f3c1b4375 --- /dev/null +++ b/tests/runtime/workingreporter/test_maxduration.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingreporter.kit import Recorder + + +class TestMaxDuration: + @pytest.mark.asyncio + async def test_keepalive_stops_after_max_but_stop_still_sends_false(self): + rec = Recorder() + reporter = WorkingStateReporter( + rec, keep_alive_seconds=0.02, max_working_state_seconds=0.05 + ) + + await reporter.start() + # Let the cap elapse and the keep-alive go quiet. + await asyncio.sleep(0.2) + count_after_cap = rec.calls.count(True) + await asyncio.sleep(0.1) + # No further true pings after the cap. + assert rec.calls.count(True) == count_after_cap + + # stop() still sends the authoritative false. + await reporter.stop() + assert rec.calls[-1] is False diff --git a/tests/runtime/workingreporter/test_ordering.py b/tests/runtime/workingreporter/test_ordering.py new file mode 100644 index 000000000..fd0edc17c --- /dev/null +++ b/tests/runtime/workingreporter/test_ordering.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingreporter.kit import Recorder + + +class TestOrdering: + @pytest.mark.asyncio + async def test_false_always_after_true_even_on_ultra_short_cycle(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) + + # Ultra-short cycle: start then immediately stop. + await reporter.start() + await reporter.stop() + + # The lock must guarantee true is sent (and lands) before false. + assert rec.calls[0] is True + assert rec.calls[-1] is False + # No false appears before the first true. + first_false = rec.calls.index(False) + first_true = rec.calls.index(True) + assert first_true < first_false diff --git a/tests/runtime/workingreporter/test_startstop.py b/tests/runtime/workingreporter/test_startstop.py new file mode 100644 index 000000000..3d87f3424 --- /dev/null +++ b/tests/runtime/workingreporter/test_startstop.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingreporter.kit import Recorder + + +class TestStartStop: + @pytest.mark.asyncio + async def test_start_sends_true_immediately(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) + + await reporter.start() + + assert rec.calls == [True] + await reporter.stop() + + @pytest.mark.asyncio + async def test_stop_sends_false(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) + + await reporter.start() + await reporter.stop() + + assert rec.calls[0] is True + assert rec.calls[-1] is False + assert rec.calls.count(False) == 1 + + @pytest.mark.asyncio + async def test_stop_without_start_is_noop(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=10.0) + + await reporter.stop() + + assert rec.calls == [] + + @pytest.mark.asyncio + async def test_disabled_reports_nothing(self): + rec = Recorder() + reporter = WorkingStateReporter(rec, keep_alive_seconds=0.01, enabled=False) + + await reporter.start() + await asyncio.sleep(0.05) + await reporter.stop() + + assert rec.calls == [] diff --git a/tests/runtime/workingreporter/test_validation.py b/tests/runtime/workingreporter/test_validation.py new file mode 100644 index 000000000..a4ce9414f --- /dev/null +++ b/tests/runtime/workingreporter/test_validation.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingreporter.kit import Recorder + + +class TestValidation: + def test_non_positive_cadence_rejected(self): + with pytest.raises(ValueError): + WorkingStateReporter(Recorder(), keep_alive_seconds=0)