From d1419b7db3c4388a746e577e9792921ae5e7d2d0 Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 10:08:37 +0300 Subject: [PATCH 1/5] fix: Keep working-state reporting alive through provider turns --- docs/turn-outcome.md | 22 +- docs/websocket-events.md | 9 +- src/band/adapters/opencode/adapter.py | 53 ++-- src/band/runtime/execution.py | 125 ++++++++-- src/band/runtime/tools/agent.py | 4 +- src/band/runtime/working_state.py | 1 + tests/runtime/activityownership/__init__.py | 1 + tests/runtime/activityownership/conftest.py | 56 +++++ tests/runtime/activityownership/kit.py | 17 ++ .../activityownership/test_lifetime.py | 198 +++++++++++++++ .../activityownership/test_opencode.py | 77 ++++++ .../runtime/executionworkingstate/__init__.py | 1 + .../runtime/executionworkingstate/conftest.py | 38 +++ tests/runtime/executionworkingstate/kit.py | 52 ++++ .../test_sessionconfigguards.py | 56 +++++ .../executionworkingstate/test_wiring.py | 104 ++++++++ tests/runtime/test_execution_working_state.py | 233 ------------------ tests/runtime/test_working_state.py | 203 --------------- tests/runtime/workingstate/__init__.py | 1 + tests/runtime/workingstate/kit.py | 28 +++ .../runtime/workingstate/test_cancellation.py | 47 ++++ .../workingstate/test_failureswallowed.py | 22 ++ tests/runtime/workingstate/test_keepalive.py | 32 +++ .../runtime/workingstate/test_maxduration.py | 29 +++ tests/runtime/workingstate/test_ordering.py | 25 ++ tests/runtime/workingstate/test_startstop.py | 52 ++++ tests/runtime/workingstate/test_validation.py | 12 + 27 files changed, 1019 insertions(+), 479 deletions(-) create mode 100644 tests/runtime/activityownership/__init__.py create mode 100644 tests/runtime/activityownership/conftest.py create mode 100644 tests/runtime/activityownership/kit.py create mode 100644 tests/runtime/activityownership/test_lifetime.py create mode 100644 tests/runtime/activityownership/test_opencode.py create mode 100644 tests/runtime/executionworkingstate/__init__.py create mode 100644 tests/runtime/executionworkingstate/conftest.py create mode 100644 tests/runtime/executionworkingstate/kit.py create mode 100644 tests/runtime/executionworkingstate/test_sessionconfigguards.py create mode 100644 tests/runtime/executionworkingstate/test_wiring.py delete mode 100644 tests/runtime/test_execution_working_state.py delete mode 100644 tests/runtime/test_working_state.py create mode 100644 tests/runtime/workingstate/__init__.py create mode 100644 tests/runtime/workingstate/kit.py create mode 100644 tests/runtime/workingstate/test_cancellation.py create mode 100644 tests/runtime/workingstate/test_failureswallowed.py create mode 100644 tests/runtime/workingstate/test_keepalive.py create mode 100644 tests/runtime/workingstate/test_maxduration.py create mode 100644 tests/runtime/workingstate/test_ordering.py create mode 100644 tests/runtime/workingstate/test_startstop.py create mode 100644 tests/runtime/workingstate/test_validation.py diff --git a/docs/turn-outcome.md b/docs/turn-outcome.md index 7402a79a1..6951fad99 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, @@ -396,3 +396,17 @@ a later existing host cleanup retry completes retirement. Worker completion alon does not readmit the room. Successful approval release retains its original tools and judgment until the worker actually finishes; `Turn.detach()` is still only an outcome-ledger operation. + +## 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 8de9c2381..a9f826d1d 100644 --- a/src/band/adapters/opencode/adapter.py +++ b/src/band/adapters/opencode/adapter.py @@ -18,6 +18,7 @@ from band.adapters.opencode.approvals import ApprovalPorts, RoomApprovals from band.adapters.opencode.config import OpencodeAdapterConfig +from band.client.streaming import ControlMode from band.converters.opencode import OpencodeHistoryConverter from band.core.delivery import relay_reply from band.core.exceptions import BandConnectionError, InputNotAccepted @@ -28,6 +29,11 @@ ) from band.core.simple_adapter import SimpleAdapter from band.core.turn import judge_detached_turn +from band.core.turn_lifecycle import ( + CYCLE_CANCEL_GRACE_SECONDS, + TurnTask, + bind_turn_task, +) from band.core.types import ( AdapterFeatures, Capability, @@ -99,6 +105,8 @@ class TurnState: assistant_part_types: dict[str, str] = field(default_factory=dict) reported_tool_calls: set[str] = field(default_factory=set) reported_tool_results: set[str] = field(default_factory=set) + worker: TurnTask | None = None + abort_started: bool = False last_error_message: str | None = None # Per-assistant-message usage for the current turn (last-write-wins per id, # since message.updated streams repeatedly). Summed across messages at turn @@ -532,6 +540,7 @@ async def on_message( turn, ) ) + turn.worker = bind_turn_task(tools, turn_task) # Register the watcher only while this turn is still current; a # superseded turn's task must not clobber (or be cancelled through) # the next turn's ambient pointer. @@ -547,21 +556,12 @@ async def on_message( # Nothing awaited here re-raises asyncio.TimeoutError, so on_message has no # timeout handler of its own. except asyncio.CancelledError: - # The runtime interrupts a turn by cancelling this coroutine, but the - # watcher runs detached. Left alone it outlives the interrupt: it - # posts the very reply the user stopped, and holds the room's busy - # guard until turn_timeout_s. Drop the turn state even if a second - # cancellation (e.g. a double-tap stop) cuts off the abandon/abort - # awaits below -- otherwise the room stays wedged on a turn whose - # state never clears. - if turn_future is not None: - assert turn is not None + if turn is not None: try: - already_aborted = await turn.approvals.abandon() + await turn.approvals.abandon() + await self._abort_turn(turn, "interrupted") finally: self._clear_turn_state(room_state, expected_turn=turn) - if not already_aborted: - await self._abort_turn(turn, "interrupted") raise except httpx.HTTPStatusError as exc: logger.exception("OpenCode request failed for room %s", room_id) @@ -1029,6 +1029,9 @@ async def _watch_turn_completion( ) try: await self._await_turn(turn) + except asyncio.CancelledError: + await self._abort_turn(turn, "interrupted") + raise except TimeoutError: logger.warning( "OpenCode turn timed out for room %s (session=%s)", @@ -1075,10 +1078,25 @@ async def _watch_turn_completion( expected_task=asyncio.current_task(), ) + async def on_interrupt(self, room_id: str, mode: ControlMode) -> None: + state = self._rooms.get(room_id) + if state is None or state.turn is None: + return + turn = state.turn + if turn.worker is not None: + turn.worker.terminate() + await turn.worker.drain() + else: + await self._abort_turn(turn, "interrupted") + async def _abort_turn(self, turn: TurnState, reason: str) -> None: """Best-effort: tell OpenCode to stop working on this room's session.""" + if turn.abort_started: + return + turn.abort_started = True try: - await turn.client.abort_session(turn.session_id) + async with asyncio.timeout(CYCLE_CANCEL_GRACE_SECONDS): + await turn.client.abort_session(turn.session_id) except Exception: logger.exception( "Failed to abort %s OpenCode session %s", @@ -1173,8 +1191,13 @@ def _clear_turn_state( return turn_task = room_state.turn_task - if turn_task is not None and turn_task is not expected_task: - turn_task.cancel() + if ( + turn_task is not None + and turn_task is not expected_task + and room_state.turn is not None + and room_state.turn.worker is not None + ): + room_state.turn.worker.terminate() if room_state.turn is not None: room_state.turn.approvals.cancel() diff --git a/src/band/runtime/execution.py b/src/band/runtime/execution.py index 6948d1133..701f25ded 100644 --- a/src/band/runtime/execution.py +++ b/src/band/runtime/execution.py @@ -413,6 +413,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 @@ -451,8 +454,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: """ @@ -629,6 +634,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 @@ -736,10 +743,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.""" @@ -767,7 +777,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 @@ -784,8 +798,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) @@ -1842,14 +1858,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 @@ -1863,8 +1884,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 @@ -1872,6 +1894,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: @@ -1957,15 +2046,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..6951d5492 100644 --- a/src/band/runtime/tools/agent.py +++ b/src/band/runtime/tools/agent.py @@ -388,8 +388,8 @@ def from_context( return tools def register_turn_task(self, owner: TurnTask) -> None: - if self._scope is not None: - self._scope.bind(owner) + if self._scope is not None and self._ctx is not None: + 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..916e0ae7b --- /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 peer() -> AsyncIterator[LifecyclePlatform]: + yield ActivityPlatform() + + +@pytest.fixture +async def execution( + peer: LifecyclePlatform, +) -> AsyncIterator[Callable[..., ExecutionContext]]: + contexts: list[ExecutionContext] = [] + async with rest_client_over(peer.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..1dcf0f921 --- /dev/null +++ b/tests/runtime/activityownership/test_lifetime.py @@ -0,0 +1,198 @@ +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], peer: 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 peer.activity_reports == [KEEP_ALIVE, KEEP_ALIVE] + assert peer.marked("processed") == ["original", "busy"] + release.set() + await workers[0].task + await asyncio.sleep(0) + await ctx._join_activity_completion() + assert peer.activity_reports[-1] == IDLE + + async def test_worker_completion_does_not_clear_running_handler( + self, execution: Callable[..., ExecutionContext], peer: 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 peer.activity_reports == [KEEP_ALIVE] + release.set() + await turn + assert peer.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], + peer: 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 peer.marked("processed") == ["original"] + assert peer.activity_reports == [KEEP_ALIVE, IDLE] + + @pytest.mark.looptime + async def test_detached_heartbeat_stop_preserves_ack( + self, execution: Callable[..., ExecutionContext], peer: 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")) + peer.stopped = True + await asyncio.sleep(3.1) + assert workers[0].task.cancelled() + assert peer.marked("processed") == ["original", "busy"] + assert peer.activity_reports[-1] == IDLE + + @pytest.mark.looptime + async def test_busy_delivery_keeps_original_cap_deadline( + self, execution: Callable[..., ExecutionContext], peer: 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 peer.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 peer.activity_reports[-1] == IDLE + + async def test_shutdown_clears_reporting_without_cancelling_detached_worker( + self, execution: Callable[..., ExecutionContext], peer: 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 peer.activity_reports == [KEEP_ALIVE, IDLE] + assert not workers[0].task.done() + release.set() + await workers[0].task + assert peer.activity_reports == [KEEP_ALIVE, IDLE] + + async def test_new_worker_admitted_after_previous_completion_gets_own_indicator( + self, execution: Callable[..., ExecutionContext], peer: 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 peer.activity_reports == [KEEP_ALIVE, IDLE, KEEP_ALIVE] + releases[1].set() + await workers[1].task + await asyncio.sleep(0) + await ctx._join_activity_completion() + assert peer.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..7fb26af66 --- /dev/null +++ b/tests/runtime/activityownership/test_opencode.py @@ -0,0 +1,77 @@ +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], + peer: 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 peer.activity_reports == [KEEP_ALIVE] + assert peer.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 ctx._join_activity_completion() + assert client.aborted_sessions == ["sess-1"] + assert cleaned.is_set() + assert peer.activity_reports == [KEEP_ALIVE, IDLE] + assert peer.marked("processed") == ["original"] + finally: + release.set() + await adapter.on_cleanup(ROOM_ID) diff --git a/tests/runtime/executionworkingstate/__init__.py b/tests/runtime/executionworkingstate/__init__.py new file mode 100644 index 000000000..9d48db4f9 --- /dev/null +++ b/tests/runtime/executionworkingstate/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/tests/runtime/executionworkingstate/conftest.py b/tests/runtime/executionworkingstate/conftest.py new file mode 100644 index 000000000..67bd0bbf7 --- /dev/null +++ b/tests/runtime/executionworkingstate/conftest.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +@pytest.fixture +def mock_link(): + link = MagicMock() + link.agent_id = "agent-123" + link.rest = MagicMock() + + participant1 = MagicMock() + participant1.id = "user-1" + participant1.name = "User One" + participant1.type = "User" + link.rest.agent_api_participants = MagicMock() + link.rest.agent_api_participants.list_agent_chat_participants = AsyncMock( + return_value=MagicMock(data=[participant1]) + ) + link.rest.agent_api_context = MagicMock() + link.rest.agent_api_context.get_agent_chat_context = AsyncMock( + return_value=MagicMock(data=[]) + ) + + link.mark_processing = AsyncMock(return_value=True) + link.mark_processed = AsyncMock(return_value=True) + link.mark_failed = AsyncMock(return_value=True) + link.get_next_message = AsyncMock(return_value=None) + link.get_stale_processing_messages = AsyncMock(return_value=[]) + link.report_activity = AsyncMock(return_value=True) + return link + + +@pytest.fixture +def mock_handler(): + return AsyncMock() diff --git a/tests/runtime/executionworkingstate/kit.py b/tests/runtime/executionworkingstate/kit.py new file mode 100644 index 000000000..0be9f6f5f --- /dev/null +++ b/tests/runtime/executionworkingstate/kit.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from unittest.mock import AsyncMock, MagicMock + +from band.runtime.messages import PlatformMessage +from band.runtime.sessionconfig import SessionConfig + + +def _working_values(link) -> list[bool]: + """Sequence of `working` booleans passed to link.report_activity.""" + return [ + c.args[1] if len(c.args) > 1 else c.kwargs["working"] + for c in link.report_activity.call_args_list + ] + + +_FAST = SessionConfig(working_keep_alive_seconds=4.0) + + +def _backlog_message( + room_id: str = "room-123", msg_id: str = "msg-b1" +) -> PlatformMessage: + return PlatformMessage( + id=msg_id, + room_id=room_id, + content="hi", + sender_id="user-1", + sender_type="User", + sender_name="User One", + message_type="text", + metadata={}, + created_at=datetime.now(UTC), + ) + + +def _independent_link(): + link = MagicMock() + link.agent_id = "agent-123" + link.rest = MagicMock() + p = MagicMock(id="user-1", name="User One", type="User") + link.rest.agent_api_participants.list_agent_chat_participants = AsyncMock( + return_value=MagicMock(data=[p]) + ) + link.rest.agent_api_context.get_agent_chat_context = AsyncMock( + return_value=MagicMock(data=[]) + ) + link.mark_processing = AsyncMock(return_value=True) + link.mark_processed = AsyncMock(return_value=True) + link.mark_failed = AsyncMock(return_value=True) + link.report_activity = AsyncMock(return_value=True) + return link diff --git a/tests/runtime/executionworkingstate/test_sessionconfigguards.py b/tests/runtime/executionworkingstate/test_sessionconfigguards.py new file mode 100644 index 000000000..8431784e0 --- /dev/null +++ b/tests/runtime/executionworkingstate/test_sessionconfigguards.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import pytest + +from band.runtime.sessionconfig import SessionConfig + + +class TestSessionConfigGuards: + def test_defaults_are_valid(self): + cfg = SessionConfig() + assert cfg.enable_working_state is True + assert cfg.working_keep_alive_seconds == 3.0 + assert cfg.working_request_timeout_seconds == 2 + assert cfg.max_working_state_seconds is None + + def test_cadence_must_be_below_ttl_half(self): + with pytest.raises(ValueError): + SessionConfig(working_keep_alive_seconds=5.0) # >= 10/2 + + def test_cadence_must_be_positive(self): + with pytest.raises(ValueError): + SessionConfig(working_keep_alive_seconds=0) + + def test_timeout_must_be_below_cadence(self): + with pytest.raises(ValueError): + SessionConfig( + working_keep_alive_seconds=3.0, working_request_timeout_seconds=3 + ) + + def test_timeout_must_be_positive(self): + with pytest.raises(ValueError): + SessionConfig(working_request_timeout_seconds=0) + + def test_max_duration_if_set_must_be_positive(self): + with pytest.raises(ValueError): + SessionConfig(max_working_state_seconds=0) + + def test_max_cycle_seconds_if_set_must_be_positive(self): + with pytest.raises(ValueError, match="max_cycle_seconds"): + SessionConfig(max_cycle_seconds=0) + with pytest.raises(ValueError, match="max_cycle_seconds"): + SessionConfig(max_cycle_seconds=-1.0) + + def test_disabled_skips_guards(self): + # When disabled, odd values must not raise. + cfg = SessionConfig( + enable_working_state=False, working_keep_alive_seconds=100.0 + ) + assert cfg.enable_working_state is False + + def test_max_cycle_seconds_guard_is_independent_of_working_state(self): + # max_cycle_seconds is a hang-killer unrelated to working-state + # reporting, so its guard must still fire even when working-state + # reporting itself is disabled. + with pytest.raises(ValueError, match="max_cycle_seconds"): + SessionConfig(enable_working_state=False, max_cycle_seconds=0) diff --git a/tests/runtime/executionworkingstate/test_wiring.py b/tests/runtime/executionworkingstate/test_wiring.py new file mode 100644 index 000000000..3ca462dfd --- /dev/null +++ b/tests/runtime/executionworkingstate/test_wiring.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest + +from band.platform.event import ReconnectedEvent +from band.runtime.execution import ExecutionContext +from band.runtime.sessionconfig import SessionConfig +from tests.conftest import ( + make_message_event, + make_participant_added_event, +) +from tests.runtime.executionworkingstate.kit import ( + _FAST, + _backlog_message, + _independent_link, + _working_values, +) + + +class TestWiring: + @pytest.mark.asyncio + async def test_message_event_reports_true_then_false(self, mock_link, mock_handler): + ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) + + await ctx._process_event(make_message_event(room_id="room-123", msg_id="m1")) + + assert _working_values(mock_link) == [True, False] + for call in mock_link.report_activity.call_args_list: + assert call.args[0] == "room-123" + + @pytest.mark.asyncio + async def test_reports_false_even_when_handler_raises(self, mock_link): + handler = AsyncMock(side_effect=RuntimeError("boom")) + ctx = ExecutionContext("room-123", mock_link, handler, config=_FAST) + + await ctx._process_event(make_message_event(room_id="room-123", msg_id="m1")) + + assert _working_values(mock_link)[-1] is False + assert _working_values(mock_link).count(False) == 1 + + @pytest.mark.asyncio + async def test_participant_event_does_not_report(self, mock_link, mock_handler): + ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) + + await ctx._process_event( + make_participant_added_event(room_id="room-123", participant_id="user-2") + ) + + mock_link.report_activity.assert_not_called() + + @pytest.mark.asyncio + async def test_reconnected_event_does_not_report(self, mock_link, mock_handler): + ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) + + await ctx._process_event(ReconnectedEvent(room_id="room-123")) + + mock_link.report_activity.assert_not_called() + + @pytest.mark.asyncio + async def test_hub_room_does_not_report(self, mock_link, mock_handler): + ctx = ExecutionContext( + "hub-1", mock_link, mock_handler, config=_FAST, hub_room_id="hub-1" + ) + + await ctx._process_event(make_message_event(room_id="hub-1", msg_id="m1")) + + mock_link.report_activity.assert_not_called() + + @pytest.mark.asyncio + async def test_backlog_message_reports_true_then_false( + self, mock_link, mock_handler + ): + ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) + + await ctx._process_backlog_message(_backlog_message()) + + assert _working_values(mock_link) == [True, False] + + @pytest.mark.asyncio + async def test_disabled_config_does_not_report(self, mock_link, mock_handler): + ctx = ExecutionContext( + "room-123", + mock_link, + mock_handler, + config=SessionConfig(enable_working_state=False), + ) + + await ctx._process_event(make_message_event(room_id="room-123", msg_id="m1")) + + mock_link.report_activity.assert_not_called() + + @pytest.mark.asyncio + async def test_rooms_are_isolated(self, mock_handler): + link_a = _independent_link() + link_b = _independent_link() + ctx_a = ExecutionContext("room-a", link_a, mock_handler, config=_FAST) + ExecutionContext("room-b", link_b, mock_handler, config=_FAST) + + await ctx_a._process_event(make_message_event(room_id="room-a", msg_id="m1")) + + link_a.report_activity.assert_called() + link_b.report_activity.assert_not_called() diff --git a/tests/runtime/test_execution_working_state.py b/tests/runtime/test_execution_working_state.py deleted file mode 100644 index fa39effdf..000000000 --- a/tests/runtime/test_execution_working_state.py +++ /dev/null @@ -1,233 +0,0 @@ -"""Integration tests: ExecutionContext wiring of the working-state reporter.""" - -from __future__ import annotations - -from datetime import UTC, datetime -from unittest.mock import AsyncMock, MagicMock - -import pytest - -from band.platform.event import ReconnectedEvent -from band.runtime.execution import ExecutionContext -from band.runtime.messages import PlatformMessage -from band.runtime.sessionconfig import SessionConfig -from tests.conftest import ( - make_message_event, - make_participant_added_event, -) - - -@pytest.fixture -def mock_link(): - link = MagicMock() - link.agent_id = "agent-123" - link.rest = MagicMock() - - participant1 = MagicMock() - participant1.id = "user-1" - participant1.name = "User One" - participant1.type = "User" - link.rest.agent_api_participants = MagicMock() - link.rest.agent_api_participants.list_agent_chat_participants = AsyncMock( - return_value=MagicMock(data=[participant1]) - ) - link.rest.agent_api_context = MagicMock() - link.rest.agent_api_context.get_agent_chat_context = AsyncMock( - return_value=MagicMock(data=[]) - ) - - link.mark_processing = AsyncMock(return_value=True) - link.mark_processed = AsyncMock(return_value=True) - link.mark_failed = AsyncMock(return_value=True) - link.get_next_message = AsyncMock(return_value=None) - link.get_stale_processing_messages = AsyncMock(return_value=[]) - link.report_activity = AsyncMock(return_value=True) - return link - - -def _working_values(link) -> list[bool]: - """Sequence of `working` booleans passed to link.report_activity.""" - return [ - c.args[1] if len(c.args) > 1 else c.kwargs["working"] - for c in link.report_activity.call_args_list - ] - - -# Keep cadence high enough that no keep-alive fires during a fast test cycle, -# but below the TTL/2 guard. -_FAST = SessionConfig(working_keep_alive_seconds=4.0) - - -def _backlog_message( - room_id: str = "room-123", msg_id: str = "msg-b1" -) -> PlatformMessage: - return PlatformMessage( - id=msg_id, - room_id=room_id, - content="hi", - sender_id="user-1", - sender_type="User", - sender_name="User One", - message_type="text", - metadata={}, - created_at=datetime.now(UTC), - ) - - -class TestWiring: - @pytest.mark.asyncio - async def test_message_event_reports_true_then_false(self, mock_link, mock_handler): - ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) - - await ctx._process_event(make_message_event(room_id="room-123", msg_id="m1")) - - assert _working_values(mock_link) == [True, False] - for call in mock_link.report_activity.call_args_list: - assert call.args[0] == "room-123" - - @pytest.mark.asyncio - async def test_reports_false_even_when_handler_raises(self, mock_link): - handler = AsyncMock(side_effect=RuntimeError("boom")) - ctx = ExecutionContext("room-123", mock_link, handler, config=_FAST) - - await ctx._process_event(make_message_event(room_id="room-123", msg_id="m1")) - - assert _working_values(mock_link)[-1] is False - assert _working_values(mock_link).count(False) == 1 - - @pytest.mark.asyncio - async def test_participant_event_does_not_report(self, mock_link, mock_handler): - ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) - - await ctx._process_event( - make_participant_added_event(room_id="room-123", participant_id="user-2") - ) - - mock_link.report_activity.assert_not_called() - - @pytest.mark.asyncio - async def test_reconnected_event_does_not_report(self, mock_link, mock_handler): - ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) - - await ctx._process_event(ReconnectedEvent(room_id="room-123")) - - mock_link.report_activity.assert_not_called() - - @pytest.mark.asyncio - async def test_hub_room_does_not_report(self, mock_link, mock_handler): - ctx = ExecutionContext( - "hub-1", mock_link, mock_handler, config=_FAST, hub_room_id="hub-1" - ) - - await ctx._process_event(make_message_event(room_id="hub-1", msg_id="m1")) - - mock_link.report_activity.assert_not_called() - - @pytest.mark.asyncio - async def test_backlog_message_reports_true_then_false( - self, mock_link, mock_handler - ): - ctx = ExecutionContext("room-123", mock_link, mock_handler, config=_FAST) - - await ctx._process_backlog_message(_backlog_message()) - - assert _working_values(mock_link) == [True, False] - - @pytest.mark.asyncio - async def test_disabled_config_does_not_report(self, mock_link, mock_handler): - ctx = ExecutionContext( - "room-123", - mock_link, - mock_handler, - config=SessionConfig(enable_working_state=False), - ) - - await ctx._process_event(make_message_event(room_id="room-123", msg_id="m1")) - - mock_link.report_activity.assert_not_called() - - @pytest.mark.asyncio - async def test_rooms_are_isolated(self, mock_handler): - link_a = _independent_link() - link_b = _independent_link() - ctx_a = ExecutionContext("room-a", link_a, mock_handler, config=_FAST) - ExecutionContext("room-b", link_b, mock_handler, config=_FAST) - - await ctx_a._process_event(make_message_event(room_id="room-a", msg_id="m1")) - - link_a.report_activity.assert_called() - link_b.report_activity.assert_not_called() - - -def _independent_link(): - link = MagicMock() - link.agent_id = "agent-123" - link.rest = MagicMock() - p = MagicMock(id="user-1", name="User One", type="User") - link.rest.agent_api_participants.list_agent_chat_participants = AsyncMock( - return_value=MagicMock(data=[p]) - ) - link.rest.agent_api_context.get_agent_chat_context = AsyncMock( - return_value=MagicMock(data=[]) - ) - link.mark_processing = AsyncMock(return_value=True) - link.mark_processed = AsyncMock(return_value=True) - link.mark_failed = AsyncMock(return_value=True) - link.report_activity = AsyncMock(return_value=True) - return link - - -@pytest.fixture -def mock_handler(): - return AsyncMock() - - -class TestSessionConfigGuards: - def test_defaults_are_valid(self): - cfg = SessionConfig() - assert cfg.enable_working_state is True - assert cfg.working_keep_alive_seconds == 3.0 - assert cfg.working_request_timeout_seconds == 2 - assert cfg.max_working_state_seconds is None - - def test_cadence_must_be_below_ttl_half(self): - with pytest.raises(ValueError): - SessionConfig(working_keep_alive_seconds=5.0) # >= 10/2 - - def test_cadence_must_be_positive(self): - with pytest.raises(ValueError): - SessionConfig(working_keep_alive_seconds=0) - - def test_timeout_must_be_below_cadence(self): - with pytest.raises(ValueError): - SessionConfig( - working_keep_alive_seconds=3.0, working_request_timeout_seconds=3 - ) - - def test_timeout_must_be_positive(self): - with pytest.raises(ValueError): - SessionConfig(working_request_timeout_seconds=0) - - def test_max_duration_if_set_must_be_positive(self): - with pytest.raises(ValueError): - SessionConfig(max_working_state_seconds=0) - - def test_max_cycle_seconds_if_set_must_be_positive(self): - with pytest.raises(ValueError, match="max_cycle_seconds"): - SessionConfig(max_cycle_seconds=0) - with pytest.raises(ValueError, match="max_cycle_seconds"): - SessionConfig(max_cycle_seconds=-1.0) - - def test_disabled_skips_guards(self): - # When disabled, odd values must not raise. - cfg = SessionConfig( - enable_working_state=False, working_keep_alive_seconds=100.0 - ) - assert cfg.enable_working_state is False - - def test_max_cycle_seconds_guard_is_independent_of_working_state(self): - # max_cycle_seconds is a hang-killer unrelated to working-state - # reporting, so its guard must still fire even when working-state - # reporting itself is disabled. - with pytest.raises(ValueError, match="max_cycle_seconds"): - SessionConfig(enable_working_state=False, max_cycle_seconds=0) 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/workingstate/__init__.py b/tests/runtime/workingstate/__init__.py new file mode 100644 index 000000000..9d48db4f9 --- /dev/null +++ b/tests/runtime/workingstate/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/tests/runtime/workingstate/kit.py b/tests/runtime/workingstate/kit.py new file mode 100644 index 000000000..ee553bf4c --- /dev/null +++ b/tests/runtime/workingstate/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/workingstate/test_cancellation.py b/tests/runtime/workingstate/test_cancellation.py new file mode 100644 index 000000000..9cb55c023 --- /dev/null +++ b/tests/runtime/workingstate/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.workingstate.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/workingstate/test_failureswallowed.py b/tests/runtime/workingstate/test_failureswallowed.py new file mode 100644 index 000000000..0dbe23eef --- /dev/null +++ b/tests/runtime/workingstate/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.workingstate.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/workingstate/test_keepalive.py b/tests/runtime/workingstate/test_keepalive.py new file mode 100644 index 000000000..ee7a06a58 --- /dev/null +++ b/tests/runtime/workingstate/test_keepalive.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingstate.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/workingstate/test_maxduration.py b/tests/runtime/workingstate/test_maxduration.py new file mode 100644 index 000000000..b16c313f4 --- /dev/null +++ b/tests/runtime/workingstate/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.workingstate.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/workingstate/test_ordering.py b/tests/runtime/workingstate/test_ordering.py new file mode 100644 index 000000000..d15ff8289 --- /dev/null +++ b/tests/runtime/workingstate/test_ordering.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingstate.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/workingstate/test_startstop.py b/tests/runtime/workingstate/test_startstop.py new file mode 100644 index 000000000..0b9f2bfb5 --- /dev/null +++ b/tests/runtime/workingstate/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.workingstate.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/workingstate/test_validation.py b/tests/runtime/workingstate/test_validation.py new file mode 100644 index 000000000..36ffbbc15 --- /dev/null +++ b/tests/runtime/workingstate/test_validation.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +import pytest + +from band.runtime.working_state import WorkingStateReporter +from tests.runtime.workingstate.kit import Recorder + + +class TestValidation: + def test_non_positive_cadence_rejected(self): + with pytest.raises(ValueError): + WorkingStateReporter(Recorder(), keep_alive_seconds=0) From 8ec26b33cf3ff15f27d6149daa27a0d4ea68c17d Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 10:25:08 +0300 Subject: [PATCH 2/5] fix: Preserve standalone worker binding across tool wrappers --- src/band/runtime/tools/agent.py | 7 +++++-- tests/runtime/helpers.py | 9 ++++++--- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/src/band/runtime/tools/agent.py b/src/band/runtime/tools/agent.py index 6951d5492..88d53634b 100644 --- a/src/band/runtime/tools/agent.py +++ b/src/band/runtime/tools/agent.py @@ -388,8 +388,11 @@ def from_context( return tools def register_turn_task(self, owner: TurnTask) -> None: - if self._scope is not None and self._ctx is not None: - self._ctx.register_turn_task(self._scope, owner) + if self._scope is not None: + 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/tests/runtime/helpers.py b/tests/runtime/helpers.py index 0a96e4340..0a04accde 100644 --- a/tests/runtime/helpers.py +++ b/tests/runtime/helpers.py @@ -296,13 +296,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) From 1476457ca6f33d8eb0f43918727d9cf16c88455b Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 11:09:08 +0300 Subject: [PATCH 3/5] test: Avoid the E2E peer fixture reservation in runtime tests --- tests/runtime/activityownership/conftest.py | 6 +-- .../activityownership/test_lifetime.py | 46 +++++++++---------- .../activityownership/test_opencode.py | 10 ++-- 3 files changed, 31 insertions(+), 31 deletions(-) diff --git a/tests/runtime/activityownership/conftest.py b/tests/runtime/activityownership/conftest.py index 916e0ae7b..16bcf6bfc 100644 --- a/tests/runtime/activityownership/conftest.py +++ b/tests/runtime/activityownership/conftest.py @@ -20,16 +20,16 @@ @pytest.fixture -async def peer() -> AsyncIterator[LifecyclePlatform]: +async def activity_platform() -> AsyncIterator[LifecyclePlatform]: yield ActivityPlatform() @pytest.fixture async def execution( - peer: LifecyclePlatform, + activity_platform: LifecyclePlatform, ) -> AsyncIterator[Callable[..., ExecutionContext]]: contexts: list[ExecutionContext] = [] - async with rest_client_over(peer.answer) as rest: + async with rest_client_over(activity_platform.answer) as rest: def create( handler: Callable[[ExecutionContext, PlatformEvent], Awaitable[None]], diff --git a/tests/runtime/activityownership/test_lifetime.py b/tests/runtime/activityownership/test_lifetime.py index 1dcf0f921..c6d8adaf5 100644 --- a/tests/runtime/activityownership/test_lifetime.py +++ b/tests/runtime/activityownership/test_lifetime.py @@ -17,7 +17,7 @@ class TestProviderLifetime: @pytest.mark.looptime async def test_detached_worker_keeps_heartbeat_across_busy_delivery( - self, execution: Callable[..., ExecutionContext], peer: LifecyclePlatform + self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform ) -> None: release = asyncio.Event() workers: list[TurnTask] = [] @@ -32,16 +32,16 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: 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 peer.activity_reports == [KEEP_ALIVE, KEEP_ALIVE] - assert peer.marked("processed") == ["original", "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 peer.activity_reports[-1] == IDLE + assert activity_platform.activity_reports[-1] == IDLE async def test_worker_completion_does_not_clear_running_handler( - self, execution: Callable[..., ExecutionContext], peer: LifecyclePlatform + self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform ) -> None: finished, release = asyncio.Event(), asyncio.Event() @@ -55,16 +55,16 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: ctx = execution(handler) turn = asyncio.create_task(ctx._process_event(make_message_event())) await finished.wait() - assert peer.activity_reports == [KEEP_ALIVE] + assert activity_platform.activity_reports == [KEEP_ALIVE] release.set() await turn - assert peer.activity_reports == [KEEP_ALIVE, IDLE] + 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], - peer: LifecyclePlatform, + activity_platform: LifecyclePlatform, mode: ControlMode, ) -> None: workers: list[TurnTask] = [] @@ -83,12 +83,12 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: await asyncio.gather(workers[0].task, return_exceptions=True) await ctx._join_activity_completion() assert workers[0].task.cancelled() - assert peer.marked("processed") == ["original"] - assert peer.activity_reports == [KEEP_ALIVE, IDLE] + 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], peer: LifecyclePlatform + self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform ) -> None: workers: list[TurnTask] = [] @@ -104,15 +104,15 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: ctx = execution(handler) await ctx._process_event(make_message_event(msg_id="original")) await ctx._process_event(make_message_event(msg_id="busy")) - peer.stopped = True + activity_platform.stopped = True await asyncio.sleep(3.1) assert workers[0].task.cancelled() - assert peer.marked("processed") == ["original", "busy"] - assert peer.activity_reports[-1] == IDLE + 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], peer: LifecyclePlatform + self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform ) -> None: release = asyncio.Event() workers: list[TurnTask] = [] @@ -131,16 +131,16 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: await asyncio.sleep(3.1) await ctx._process_event(make_message_event(msg_id="busy")) await asyncio.sleep(6) - assert peer.activity_reports == [KEEP_ALIVE, KEEP_ALIVE] + 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 peer.activity_reports[-1] == IDLE + assert activity_platform.activity_reports[-1] == IDLE async def test_shutdown_clears_reporting_without_cancelling_detached_worker( - self, execution: Callable[..., ExecutionContext], peer: LifecyclePlatform + self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform ) -> None: release = asyncio.Event() workers: list[TurnTask] = [] @@ -155,14 +155,14 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: ctx = execution(handler) await ctx._process_event(make_message_event()) await ctx.stop() - assert peer.activity_reports == [KEEP_ALIVE, IDLE] + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] assert not workers[0].task.done() release.set() await workers[0].task - assert peer.activity_reports == [KEEP_ALIVE, IDLE] + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] async def test_new_worker_admitted_after_previous_completion_gets_own_indicator( - self, execution: Callable[..., ExecutionContext], peer: LifecyclePlatform + self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform ) -> None: releases = [asyncio.Event(), asyncio.Event()] entered, admitted = asyncio.Event(), asyncio.Event() @@ -190,9 +190,9 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: await ctx._join_activity_completion() admitted.set() await turn - assert peer.activity_reports == [KEEP_ALIVE, IDLE, KEEP_ALIVE] + 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 peer.activity_reports == [KEEP_ALIVE, IDLE, KEEP_ALIVE, IDLE] + 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 index 7fb26af66..7d8e09879 100644 --- a/tests/runtime/activityownership/test_opencode.py +++ b/tests/runtime/activityownership/test_opencode.py @@ -27,7 +27,7 @@ class TestOpenCodeLifetime: async def test_parked_native_turn_dual_dispatch_aborts_remote_once( self, execution: Callable[..., ExecutionContext], - peer: LifecyclePlatform, + activity_platform: LifecyclePlatform, mode: ControlMode, ) -> None: client = FakeOpencodeClient( @@ -59,8 +59,8 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: ctx = execution(handler) try: await ctx._process_event(make_message_event(msg_id="original")) - assert peer.activity_reports == [KEEP_ALIVE] - assert peer.marked("processed") == ["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() @@ -70,8 +70,8 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: await ctx._join_activity_completion() assert client.aborted_sessions == ["sess-1"] assert cleaned.is_set() - assert peer.activity_reports == [KEEP_ALIVE, IDLE] - assert peer.marked("processed") == ["original"] + assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE] + assert activity_platform.marked("processed") == ["original"] finally: release.set() await adapter.on_cleanup(ROOM_ID) From 8bfeac07a5dc34eaf450fd6b1b3f418a9f41c8e8 Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 11:11:32 +0300 Subject: [PATCH 4/5] style: Format the renamed activity platform fixture --- .../activityownership/test_lifetime.py | 31 ++++++++++++++----- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/tests/runtime/activityownership/test_lifetime.py b/tests/runtime/activityownership/test_lifetime.py index c6d8adaf5..225251892 100644 --- a/tests/runtime/activityownership/test_lifetime.py +++ b/tests/runtime/activityownership/test_lifetime.py @@ -17,7 +17,9 @@ class TestProviderLifetime: @pytest.mark.looptime async def test_detached_worker_keeps_heartbeat_across_busy_delivery( - self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, ) -> None: release = asyncio.Event() workers: list[TurnTask] = [] @@ -41,7 +43,9 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: assert activity_platform.activity_reports[-1] == IDLE async def test_worker_completion_does_not_clear_running_handler( - self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, ) -> None: finished, release = asyncio.Event(), asyncio.Event() @@ -88,7 +92,9 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: @pytest.mark.looptime async def test_detached_heartbeat_stop_preserves_ack( - self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, ) -> None: workers: list[TurnTask] = [] @@ -112,7 +118,9 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: @pytest.mark.looptime async def test_busy_delivery_keeps_original_cap_deadline( - self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, ) -> None: release = asyncio.Event() workers: list[TurnTask] = [] @@ -140,7 +148,9 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: assert activity_platform.activity_reports[-1] == IDLE async def test_shutdown_clears_reporting_without_cancelling_detached_worker( - self, execution: Callable[..., ExecutionContext], activity_platform: LifecyclePlatform + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, ) -> None: release = asyncio.Event() workers: list[TurnTask] = [] @@ -162,7 +172,9 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: 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 + self, + execution: Callable[..., ExecutionContext], + activity_platform: LifecyclePlatform, ) -> None: releases = [asyncio.Event(), asyncio.Event()] entered, admitted = asyncio.Event(), asyncio.Event() @@ -195,4 +207,9 @@ async def handler(ctx: ExecutionContext, event: PlatformEvent) -> None: await workers[1].task await asyncio.sleep(0) await ctx._join_activity_completion() - assert activity_platform.activity_reports == [KEEP_ALIVE, IDLE, KEEP_ALIVE, IDLE] + assert activity_platform.activity_reports == [ + KEEP_ALIVE, + IDLE, + KEEP_ALIVE, + IDLE, + ] From a397d8cb905d5316a87f62da23c8c5af1895e78e Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 14:03:47 +0300 Subject: [PATCH 5/5] test: Use real time for randomized scheduler grace controls --- tests/runtime/turnadmission/test_grace_controls.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/runtime/turnadmission/test_grace_controls.py b/tests/runtime/turnadmission/test_grace_controls.py index 42bc64fdd..0e2e4f44b 100644 --- a/tests/runtime/turnadmission/test_grace_controls.py +++ b/tests/runtime/turnadmission/test_grace_controls.py @@ -24,8 +24,8 @@ WITHIN_GRACE = CYCLE_CANCEL_GRACE_SECONDS / 2 +# Real time avoids looptime rounding randomized reconciliation deadlines down. @pytest.mark.timeout(10) -@pytest.mark.looptime class TestGraceControls: """STOP during the platform-stop observer's grace leaves the next turn unstarted."""