From 734d9556a8306ea6d2dd4a4dab388d515a831fd8 Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 07:01:50 +0300 Subject: [PATCH 1/3] fix: stop Parlant session processing and preamble generation --- .github/workflows/ci.yml | 1 + src/band/adapters/parlant/adapter.py | 94 ++++++++++-- src/band/adapters/parlant/messages.py | 46 ++++++ src/band/adapters/parlant/responses.py | 23 +-- tests/adapters/parlant/conftest.py | 37 ++++- tests/adapters/parlant/helpers.py | 10 ++ .../stopprobes/parlant.py | 142 ++++++++++++++++++ .../test_backend_stop.py | 47 ++++-- .../test_parlant_job_coverage.py | 1 + tests/integrations/parlant/stubnlp.py | 97 ++++++++++++ .../integrations/parlant/test_preambletask.py | 26 ++++ 11 files changed, 489 insertions(+), 35 deletions(-) create mode 100644 tests/framework_conformance/stopprobes/parlant.py create mode 100644 tests/integrations/parlant/stubnlp.py create mode 100644 tests/integrations/parlant/test_preambletask.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 15b59af2d..c66caae0e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -317,6 +317,7 @@ jobs: tests/integrations/parlant/test_customtools.py \ tests/integrations/parlant/test_filetools.py \ tests/integrations/parlant/test_guard.py \ + tests/integrations/parlant/test_preambletask.py \ tests/integrations/parlant/test_selection.py \ tests/integrations/parlant/test_server.py \ tests/integrations/parlant/test_tasktools.py \ diff --git a/src/band/adapters/parlant/adapter.py b/src/band/adapters/parlant/adapter.py index 8dcc26b1b..3d489c6d6 100644 --- a/src/band/adapters/parlant/adapter.py +++ b/src/band/adapters/parlant/adapter.py @@ -9,6 +9,7 @@ import logging from collections.abc import Sequence +from dataclasses import dataclass from typing import TYPE_CHECKING, Any, ClassVar, NoReturn from band_sdk_core import AgentFailure @@ -18,13 +19,20 @@ from band.adapters.parlant.guidelines import GuidelineLedger, GuidelineSpec from band.adapters.parlant.history import inject_history, name_this_agent_in_history from band.adapters.parlant.lifecycle import ServerLifecycle -from band.adapters.parlant.messages import compose_user_message, post_user_message +from band.adapters.parlant.messages import ( + await_stopped_task, + compose_user_message, + live_preamble_tasks, + post_user_message, + process_session_tag, +) from band.adapters.parlant.responses import relay_agent_response from band.adapters.parlant.rooms import RoomSessions from band.converters.parlant import ParlantHistoryConverter, ParlantMessages from band.core.delivery import DeliveryFailedError, reraise_delivery_cause from band.core.protocols import GENERIC_PROVIDER_FAILURE_MESSAGE, AgentToolsProtocol from band.core.simple_adapter import SimpleAdapter +from band.core.turn_lifecycle import BackendTurns from band.core.types import Capability, Emit, FeatureKwargs, PlatformMessage from band.integrations.parlant.customschema import check_parlant_custom_tools from band.integrations.parlant.sessiontools import bound_session_tools @@ -34,6 +42,7 @@ if TYPE_CHECKING: import parlant.sdk as p from parlant.core.application import Application + from parlant.core.background_tasks import BackgroundTaskService from parlant.core.sessions import SessionId logger = logging.getLogger(__name__) @@ -49,6 +58,17 @@ "in configure=." ) +# Recorded by Parlant on the cancelled session task. +STOP_REASON = "Band turn stopped" + + +@dataclass(frozen=True) +class ParlantSessionTurn: + """The engine's background processing of one session's response.""" + + background_tasks: BackgroundTaskService + session_id: SessionId + class ParlantAdapter(SimpleAdapter[ParlantMessages]): """ @@ -165,7 +185,11 @@ def __init__( self._guidelines = GuidelineLedger() self._tools: list[Any] = [] self._rooms: RoomSessions | None = None + self._background_tasks: BackgroundTaskService | None = None self._started = False + self._backend_turns = BackendTurns[ParlantSessionTurn]( + self._cancel_processing, provider=PROVIDER + ) def add_guideline( self, @@ -284,7 +308,11 @@ async def _prepare_server( await self._configure(server, agent) from parlant.core.application import Application # noqa: PLC0415 + from parlant.core.background_tasks import ( # noqa: PLC0415 + BackgroundTaskService, + ) + self._background_tasks = server.container[BackgroundTaskService] return agent, server.container[Application] def _warn_if_custom_tools_unreachable(self) -> None: @@ -305,6 +333,10 @@ def _forget_guidelines_if_agent_gone(self) -> None: if self._lifecycle.agent is None: self._guidelines.forget_applied() + @property + def backend_turns(self) -> BackendTurns[ParlantSessionTurn]: + return self._backend_turns + async def on_message( self, msg: PlatformMessage, @@ -410,30 +442,64 @@ async def _converse( tools: AgentToolsProtocol, sender_name: str, ) -> None: - """Post *user_message* as the customer, then relay the engine's reply.""" - offset = await post_user_message( - app=app, session_id=session_id, message=user_message - ) - await relay_agent_response( - app=app, - session_id=session_id, - room_id=room_id, - min_offset=offset, - tools=tools, - sender_name=sender_name, - timeout=self.config.response_timeout, - poll=self.config.response_poll, + """Post *user_message* as the customer, then relay the engine's reply. + + The engine answers in its own background task, so ending this turn + early cancels that task too. + """ + handle = ( + None + if self._background_tasks is None + else ParlantSessionTurn(self._background_tasks, session_id) ) + async with self.backend_turns.running(room_id, handle): + offset = await post_user_message( + app=app, session_id=session_id, message=user_message + ) + gave_up = await relay_agent_response( + app=app, + session_id=session_id, + room_id=room_id, + min_offset=offset, + tools=tools, + sender_name=sender_name, + timeout=self.config.response_timeout, + poll=self.config.response_poll, + ) + if gave_up: + await self.backend_turns.stop(room_id) + + @staticmethod + async def _cancel_processing(turn: ParlantSessionTurn) -> None: + """Cancel the session's process-session task and its preamble child. + + Parlant 3.3.2's cancel marks the parent and returns without waiting, + and the parent does not cancel the preamble task it already created. + Waiting for the parent first is what makes that child visible. + """ + tag = process_session_tag(turn.session_id) + parent = turn.background_tasks._tasks.get(tag) + await turn.background_tasks.cancel(tag=tag, reason=STOP_REASON) + if parent is not None: + await await_stopped_task(parent) + children = live_preamble_tasks(str(turn.session_id)) + for child in children: + child.cancel() + for child in children: + await await_stopped_task(child) async def on_cleanup(self, room_id: str) -> None: """Forget the room's Parlant session when the agent leaves it.""" + await self.backend_turns.settled(room_id) if self._rooms is not None: self._rooms.forget(room_id) logger.debug("Room %s: Cleaned up Parlant session", room_id) async def cleanup_all(self) -> None: """Release all sessions and the owned Parlant server (call on stop).""" + await self.backend_turns.drain() self._rooms = None + self._background_tasks = None await self._lifecycle.release() self._forget_guidelines_if_agent_gone() self._started = False diff --git a/src/band/adapters/parlant/messages.py b/src/band/adapters/parlant/messages.py index e911dd0a0..86c0f6cb0 100644 --- a/src/band/adapters/parlant/messages.py +++ b/src/band/adapters/parlant/messages.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import re from dataclasses import replace from typing import TYPE_CHECKING, Any @@ -71,6 +72,51 @@ def compose_user_message( return user_message +def process_session_tag(session_id: SessionId | str) -> str: + """The tag Parlant runs a session's response under in its background tasks. + + Mirrors Parlant's own tag (pinned by a parity test against the engine), so + cancelling it stops the engine's work on the session. + """ + return f"process-session({session_id})" + + +# Parlant 3.3.2 creates this child with asyncio.create_task inside +# AlphaEngine._get_preamble_task and does not cancel it when process-session +# is cancelled. The nested coroutine's qualname is the class that defines it. +PREAMBLE_TASK_QUALNAME = "AlphaEngine._get_preamble_task..preamble_task" + + +def live_preamble_tasks(session_id: str) -> list[asyncio.Task[Any]]: + """Preamble generation still running for ``session_id``.""" + tasks: list[asyncio.Task[Any]] = [] + for task in asyncio.all_tasks(): + if task.done(): + continue + frame = next( + ( + suspended + for suspended in task.get_stack() + if suspended.f_code.co_qualname == PREAMBLE_TASK_QUALNAME + ), + None, + ) + if frame is None: + continue + context = frame.f_locals.get("context") + session = getattr(context, "session", None) + if str(getattr(session, "id", "")) == str(session_id): + tasks.append(task) + return tasks + + +async def await_stopped_task(task: asyncio.Task[Any]) -> None: + """Wait until ``task`` finishes, including when it was cancelled.""" + await asyncio.wait({task}) + if not task.cancelled(): + task.exception() + + async def post_user_message( *, app: Application, session_id: SessionId, message: str ) -> int: diff --git a/src/band/adapters/parlant/responses.py b/src/band/adapters/parlant/responses.py index fbdfccacc..4ddc8dfaf 100644 --- a/src/band/adapters/parlant/responses.py +++ b/src/band/adapters/parlant/responses.py @@ -119,13 +119,16 @@ async def relay_agent_response( sender_name: str, timeout: float, poll: float, -) -> None: +) -> bool: """Relay the engine's final reply to *sender_name*, waiting up to *timeout*. Waiting polls in *poll*-second windows and keeps going past empty ones, so a slow (cold-start) turn is still answered. If the budget elapses with no final message the turn is given up honestly: Parlant intermittently stalls after a preamble, and a preamble is not an answer. + + Returns whether the turn was given up that way, with the engine possibly + still working on it. """ offset = min_offset # perf_counter keeps its resolution on Windows, where monotonic() is coarse. @@ -144,13 +147,13 @@ async def relay_agent_response( "Room %s: stopped waiting for a Parlant reply because reading events failed", room_id, ) - return + return False case PollMiss.NO_UPDATE if tools.turn.replied: logger.info( "Room %s: stopped waiting for a Parlant reply because the turn already replied or declined", room_id, ) - return + return False case PollMiss(): await _back_off(deadline) continue @@ -169,15 +172,17 @@ async def relay_agent_response( room_id, final_text, ) - return + return False if batch.final_seen: - return + return False logger.debug("Room %s: only a preamble so far; still waiting", room_id) - if not tools.turn.replied: - logger.warning( - "Room %s: timed out after %ss waiting for agent response", room_id, timeout - ) + if tools.turn.replied: + return False + logger.warning( + "Room %s: timed out after %ss waiting for agent response", room_id, timeout + ) + return True def _message_parts(data: Any) -> tuple[str, list[str]]: diff --git a/tests/adapters/parlant/conftest.py b/tests/adapters/parlant/conftest.py index 54833aa6f..772534584 100644 --- a/tests/adapters/parlant/conftest.py +++ b/tests/adapters/parlant/conftest.py @@ -23,6 +23,7 @@ MESSAGE_KIND, SENDER_NAME, SESSION_ID, + RecordingBackgroundTasks, ) from tests.integrations.parlant.sessions import ( clear_session_tools, # noqa: F401 -- autouse fixture @@ -77,6 +78,27 @@ def application_class(mock_app: MagicMock) -> Iterator[MagicMock]: yield application +@pytest.fixture +def background_tasks() -> RecordingBackgroundTasks: + """The engine's background-task service, recording what it cancels.""" + return RecordingBackgroundTasks() + + +@pytest.fixture +def background_task_service_class() -> Iterator[MagicMock]: + """Fake ``parlant.core.background_tasks`` so on_started's import resolves.""" + service_class = MagicMock(name="BackgroundTaskService") + with patch.dict( + sys.modules, + { + "parlant.core.background_tasks": MagicMock( + BackgroundTaskService=service_class + ) + }, + ): + yield service_class + + @pytest.fixture def parlant_sessions() -> Iterator[None]: """Fake the Parlant session modules a turn imports.""" @@ -97,10 +119,19 @@ def parlant_sessions() -> Iterator[None]: @pytest.fixture -def mock_parlant_server(mock_app: MagicMock, application_class: MagicMock) -> MagicMock: - """A Parlant server whose container holds ``mock_app``.""" +def mock_parlant_server( + mock_app: MagicMock, + application_class: MagicMock, + background_tasks: RecordingBackgroundTasks, + background_task_service_class: MagicMock, +) -> MagicMock: + """A Parlant server whose container holds ``mock_app`` and the recording + background-task service.""" server = MagicMock() - server.container = {application_class: mock_app} + server.container = { + application_class: mock_app, + background_task_service_class: background_tasks, + } server.create_customer = AsyncMock(return_value=MagicMock(id="customer-123")) created_agent = MagicMock() created_agent.id = "parlant-agent-created" diff --git a/tests/adapters/parlant/helpers.py b/tests/adapters/parlant/helpers.py index 1407f3412..dd4f0f69c 100644 --- a/tests/adapters/parlant/helpers.py +++ b/tests/adapters/parlant/helpers.py @@ -39,3 +39,13 @@ def agent_event( event.offset = offset event.data = {"message": message, "tags": tags or []} return event + + +class RecordingBackgroundTasks: + """Parlant's background-task service, recording each cancelled tag.""" + + def __init__(self) -> None: + self.cancelled: list[str] = [] + + async def cancel(self, *, tag: str, reason: str = "(not given)") -> None: + self.cancelled.append(tag) diff --git a/tests/framework_conformance/stopprobes/parlant.py b/tests/framework_conformance/stopprobes/parlant.py new file mode 100644 index 000000000..6454e2a2e --- /dev/null +++ b/tests/framework_conformance/stopprobes/parlant.py @@ -0,0 +1,142 @@ +"""Backend-stop probes for the Parlant adapter on the real in-process engine; +see ``stopprobes``. + +The engine answers a session from its own background task, and the stub model +never answers, so that task ends only when it is stopped. The evidence is that +the task ``process_session_tag`` names finishes after the stop, and that the +preamble child captured before the stop is done, with nothing posted. Every +case skips outside the dev-parlant venv. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any + +from band.adapters.parlant import ParlantAdapter, ParlantAdapterConfig +from band.adapters.parlant.messages import live_preamble_tasks, process_session_tag +from band.client.streaming import ControlMode +from band.testing.fake_tools import FakeAgentTools +from tests.framework_conformance.stopprobes import ( + ENDING_CONTROLS, + StopCase, + end_turn, + stop_room, + wait_until, +) +from tests.framework_conformance.turnprobes import ROOM_ID, turn_input + +LOOPBACK = "127.0.0.1" +# Real engine work (startup aside) settles well within this, and it stays +# under the suite's 30s test timeout so a turn that never stops fails here. +ENGINE_WAIT_SECONDS = 15.0 +ENGINE_STOPPED = "engine stopped processing the session" +PREAMBLE_STOPPED = "preamble generation stopped" +TAG_IS_LIVE = "process_session_tag names the engine's running task" + + +@asynccontextmanager +async def started() -> AsyncIterator[tuple[ParlantAdapter, Any]]: + """A started adapter on an owned server whose model never answers.""" + import parlant.sdk as p # noqa: PLC0415 -- dev-parlant extra + + from tests.integrations.parlant.stubnlp import ( # noqa: PLC0415 -- dev-parlant extra + StubNLP, + ) + + nlp = StubNLP() + adapter = ParlantAdapter( + ParlantAdapterConfig(), + nlp_service=nlp.factory, + server_options={"host": LOOPBACK, "log_level": p.LogLevel.WARNING}, + ) + await adapter.on_started("Agent", "A Parlant agent") + try: + yield adapter, nlp + finally: + await adapter.cleanup_all() + + +def background_tasks(adapter: ParlantAdapter) -> Any: + from parlant.core.background_tasks import ( # noqa: PLC0415 -- dev-parlant extra + BackgroundTaskService, + ) + + return adapter.server.container[BackgroundTaskService] + + +async def processing_turn( + adapter: ParlantAdapter, nlp: Any, tools: FakeAgentTools +) -> tuple[asyncio.Task[None], str]: + """A turn the engine is generating for, and its session id.""" + turn = asyncio.create_task(adapter.on_event(turn_input(tools))) + await asyncio.wait_for(nlp.entered.wait(), ENGINE_WAIT_SECONDS) + assert adapter._rooms is not None + session_id = await adapter._rooms.session_for(ROOM_ID, customer_name="") + return turn, str(session_id) + + +async def end_running_turn(mode: ControlMode) -> list[str]: + async with started() as (adapter, nlp): + tools = FakeAgentTools() + turn, session_id = await processing_turn(adapter, nlp, tools) + engine = background_tasks(adapter)._tasks[process_session_tag(session_id)] + preamble: list[asyncio.Task[Any]] = [] + + def preamble_is_running() -> bool: + preamble[:] = live_preamble_tasks(session_id) + return bool(preamble) + + await wait_until(preamble_is_running) + await end_turn(adapter, turn, ROOM_ID, mode) + stopped, _ = await asyncio.wait({engine}, timeout=ENGINE_WAIT_SECONDS) + preamble_stopped = bool(preamble) and all(task.done() for task in preamble) + return [ + *([ENGINE_STOPPED] if stopped else []), + *([PREAMBLE_STOPPED] if preamble_stopped else []), + *(f"posted {sent}" for sent in tools.messages_sent), + ] + + +async def tag_parity() -> list[str]: + async with started() as (adapter, nlp): + turn, session_id = await processing_turn(adapter, nlp, FakeAgentTools()) + live = process_session_tag(session_id) in background_tasks(adapter)._tasks + await end_turn(adapter, turn, ROOM_ID, ControlMode.STOP) + return [TAG_IS_LIVE] if live else [] + + +def record_cancels(adapter: ParlantAdapter) -> list[str]: + """Record every tag the engine is asked to cancel, still cancelling it.""" + service = background_tasks(adapter) + cancel = service.cancel + cancelled: list[str] = [] + + async def recording(*, tag: str, reason: str = "(not given)") -> None: + cancelled.append(tag) + await cancel(tag=tag, reason=reason) + + service.cancel = recording + return cancelled + + +async def idle() -> list[str]: + async with started() as (adapter, _nlp): + cancelled = record_cancels(adapter) + await stop_room(adapter, ROOM_ID) + return cancelled + + +CASES: dict[str, StopCase] = { + **{ + f"{mode}-mid-turn": StopCase( + run=lambda mode=mode: end_running_turn(mode), + expected=[ENGINE_STOPPED, PREAMBLE_STOPPED], + ) + for mode in ENDING_CONTROLS + }, + "tag-matches-the-engine": StopCase(run=tag_parity, expected=[TAG_IS_LIVE]), + "idle-room": StopCase(run=idle, expected=[]), +} diff --git a/tests/framework_conformance/test_backend_stop.py b/tests/framework_conformance/test_backend_stop.py index 7c4efe781..caa4993ca 100644 --- a/tests/framework_conformance/test_backend_stop.py +++ b/tests/framework_conformance/test_backend_stop.py @@ -1,12 +1,14 @@ -"""Adapters using the shared cancellation owner stop their backend turns.""" +"""Every adapter that drives a backend stops that backend's turn when the +room's turn ends early; every other adapter states why it has none.""" from __future__ import annotations import pytest +import band.adapters from band.core.simple_adapter import SimpleAdapter from tests.framework_conformance.outcomeprobes import band_adapters -from tests.framework_conformance.stopprobes import StopCase +from tests.framework_conformance.stopprobes import NoBackendTurn, StopCase from tests.framework_conformance.stopprobes import a2a as a2a_probe from tests.framework_conformance.stopprobes import acp as acp_probe from tests.framework_conformance.stopprobes import claudesdk as claudesdk_probe @@ -15,6 +17,7 @@ from tests.framework_conformance.stopprobes import cursoracp as cursoracp_probe from tests.framework_conformance.stopprobes import letta as letta_probe from tests.framework_conformance.stopprobes import opencode as opencode_probe +from tests.framework_conformance.stopprobes import parlant as parlant_probe #: The stop cases of each adapter that drives a backend, by class name. STOP_PROBES: dict[str, dict[str, StopCase]] = { @@ -26,6 +29,7 @@ "CursorACPAdapter": cursoracp_probe.CASES, "LettaAdapter": letta_probe.CASES, "OpencodeAdapter": opencode_probe.CASES, + "ParlantAdapter": parlant_probe.CASES, } #: Subclasses that keep a probed parent's ``backend_turns`` and ``on_message``, @@ -35,6 +39,23 @@ "OmpACPAdapter": "ACPClientAdapter", } +#: Adapters with no backend turn to stop, and why. +NO_BACKEND_TURN: dict[str, NoBackendTurn] = { + "AgnoAdapter": NoBackendTurn.IN_PROCESS, + "AnthropicAdapter": NoBackendTurn.IN_PROCESS, + "CrewAIAdapter": NoBackendTurn.IN_PROCESS, + "CrewAIFlowAdapter": NoBackendTurn.IN_PROCESS, + "GeminiAdapter": NoBackendTurn.IN_PROCESS, + "GoogleADKAdapter": NoBackendTurn.IN_PROCESS, + "LangGraphAdapter": NoBackendTurn.IN_PROCESS, + "NooaAdapter": NoBackendTurn.IN_PROCESS, + "PydanticAIAdapter": NoBackendTurn.IN_PROCESS, + "StrandsAdapter": NoBackendTurn.IN_PROCESS, + "A2AGatewayAdapter": NoBackendTurn.GATEWAY, + "BandACPServerAdapter": NoBackendTurn.GATEWAY, + "SlackAdapter": NoBackendTurn.CHANNEL, +} + #: The module each probe needs beyond the dev venv's always-installed core; #: a venv without it skips that probe's cases. REQUIRED_MODULES: dict[str, str] = { @@ -43,6 +64,7 @@ "CopilotSDKAdapter": "copilot", "CursorACPAdapter": "acp", "LettaAdapter": "letta_client", + "ParlantAdapter": "parlant.sdk", } CASES = [ @@ -52,6 +74,9 @@ ] +ACCOUNTED = STOP_PROBES.keys() | INHERITED_PROBES.keys() | NO_BACKEND_TURN.keys() + + def runs_parents_stop(cls: type[SimpleAdapter], parent: type[SimpleAdapter]) -> bool: return ( issubclass(cls, parent) @@ -61,14 +86,18 @@ def runs_parents_stop(cls: type[SimpleAdapter], parent: type[SimpleAdapter]) -> class TestBackendStop: - def test_every_migrated_adapter_is_probed(self) -> None: - migrated = { - name - for name, cls in band_adapters().items() - if cls.backend_turns is not SimpleAdapter.backend_turns - } + def test_every_band_adapter_is_probed_or_pinned(self) -> None: + adapters = band_adapters() + + assert adapters.keys() - ACCOUNTED == set() + assert not STOP_PROBES.keys() & NO_BACKEND_TURN.keys() + + def test_every_exported_adapter_is_probed_or_pinned_even_without_its_extra( + self, + ) -> None: + exported = {name for name in band.adapters.__all__ if name.endswith("Adapter")} - assert migrated - (STOP_PROBES.keys() | INHERITED_PROBES.keys()) == set() + assert exported - ACCOUNTED == set() def test_inherited_probes_really_run_the_parents_stop(self) -> None: adapters = band_adapters() diff --git a/tests/framework_conformance/test_parlant_job_coverage.py b/tests/framework_conformance/test_parlant_job_coverage.py index 3d68b9d53..b0bf342e3 100644 --- a/tests/framework_conformance/test_parlant_job_coverage.py +++ b/tests/framework_conformance/test_parlant_job_coverage.py @@ -66,6 +66,7 @@ def test_detection_finds_the_tests_that_only_the_parlant_venv_can_run(self) -> N Path("tests/integrations/parlant/test_customtools.py"), Path("tests/integrations/parlant/test_filetools.py"), Path("tests/integrations/parlant/test_guard.py"), + Path("tests/integrations/parlant/test_preambletask.py"), Path("tests/integrations/parlant/test_selection.py"), Path("tests/integrations/parlant/test_server.py"), Path("tests/integrations/parlant/test_tasktools.py"), diff --git a/tests/integrations/parlant/stubnlp.py b/tests/integrations/parlant/stubnlp.py new file mode 100644 index 000000000..a92f9f349 --- /dev/null +++ b/tests/integrations/parlant/stubnlp.py @@ -0,0 +1,97 @@ +"""A Parlant NLP service that never answers, for driving the real engine +without a model: every generation parks until the engine cancels it. + +Import only where ``parlant`` is installed (the dev-parlant venv). +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Mapping +from typing import Any, Generic, TypeVar + +from parlant.core.nlp.embedding import Embedder, EmbeddingResult +from parlant.core.nlp.generation import SchematicGenerator +from parlant.core.nlp.moderation import ModerationService, NoModeration +from parlant.core.nlp.service import NLPService +from parlant.core.nlp.tokenization import EstimatingTokenizer, ZeroEstimatingTokenizer + +T = TypeVar("T") + +EMBEDDING_DIMENSIONS = 8 +MAX_TOKENS = 100_000 + + +class ParkedGenerator(SchematicGenerator[T], Generic[T]): + """Every generation parks, so a turn stays in the engine until cancelled.""" + + def __init__(self, entered: asyncio.Event) -> None: + self._entered = entered + + async def generate(self, prompt: Any, hints: Mapping[str, Any] = {}) -> Any: + self._entered.set() + await asyncio.Future[None]() + + @property + def id(self) -> str: + return "parked" + + @property + def max_tokens(self) -> int: + return MAX_TOKENS + + @property + def tokenizer(self) -> EstimatingTokenizer: + return ZeroEstimatingTokenizer() + + +class ZeroEmbedder(Embedder): + async def embed( + self, texts: list[str], hints: Mapping[str, Any] = {} + ) -> EmbeddingResult: + return EmbeddingResult(vectors=[[0.0] * EMBEDDING_DIMENSIONS for _ in texts]) + + @property + def id(self) -> str: + return "zero" + + @property + def max_tokens(self) -> int: + return MAX_TOKENS + + @property + def tokenizer(self) -> EstimatingTokenizer: + return ZeroEstimatingTokenizer() + + @property + def dimensions(self) -> int: + return EMBEDDING_DIMENSIONS + + +class StubNLP(NLPService): + """``entered`` is set once the engine asks the model anything.""" + + def __init__(self) -> None: + self.entered = asyncio.Event() + + def factory(self, container: Any) -> NLPService: + """The ``nlp_service=`` callable a Parlant server takes.""" + return self + + @property + def supports_streaming(self) -> bool: + return False + + async def get_schematic_generator( + self, t: type[T], hints: Mapping[str, Any] = {} + ) -> SchematicGenerator[T]: + return ParkedGenerator[t](self.entered) + + async def get_streaming_text_generator(self, hints: Mapping[str, Any] = {}) -> Any: + raise NotImplementedError + + async def get_embedder(self, hints: Mapping[str, Any] = {}) -> Embedder: + return ZeroEmbedder() + + async def get_moderation_service(self) -> ModerationService: + return NoModeration() diff --git a/tests/integrations/parlant/test_preambletask.py b/tests/integrations/parlant/test_preambletask.py new file mode 100644 index 000000000..d3f4ccb25 --- /dev/null +++ b/tests/integrations/parlant/test_preambletask.py @@ -0,0 +1,26 @@ +"""The copied preamble-task qualname matches the coroutine Parlant compiles. + +Parlant 3.3.2 defines that child on ``AlphaEngine``, so a copy that names +``Engine`` never finds the running task. This skips outside the dev-parlant venv. +""" + +from __future__ import annotations + +import types + +import pytest + +from band.adapters.parlant.messages import PREAMBLE_TASK_QUALNAME + +engine = pytest.importorskip("parlant.core.engines.alpha.engine") + + +class TestPreambleTaskIdentity: + def test_qualname_matches_the_installed_coroutine(self) -> None: + nested = [ + const.co_qualname + for const in engine.AlphaEngine._get_preamble_task.__code__.co_consts + if isinstance(const, types.CodeType) and const.co_name == "preamble_task" + ] + + assert nested == [PREAMBLE_TASK_QUALNAME] From d9fc6850c3843ad45a6560ed01367fbb3df05a44 Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 08:49:06 +0300 Subject: [PATCH 2/3] fix: test Parlant cancellation against one virtual clock --- src/band/adapters/parlant/responses.py | 11 +- tests/adapters/parlant/conftest.py | 25 +++- tests/adapters/parlant/helpers.py | 17 +++ tests/adapters/parlant/test_timeout.py | 30 +++++ .../stopprobes/__init__.py | 3 + .../stopprobes/parlant.py | 115 +++++++++++++++--- .../test_backend_stop.py | 2 +- tests/integrations/parlant/stubnlp.py | 19 ++- 8 files changed, 194 insertions(+), 28 deletions(-) create mode 100644 tests/adapters/parlant/test_timeout.py diff --git a/src/band/adapters/parlant/responses.py b/src/band/adapters/parlant/responses.py index 4ddc8dfaf..27340a892 100644 --- a/src/band/adapters/parlant/responses.py +++ b/src/band/adapters/parlant/responses.py @@ -11,7 +11,6 @@ import asyncio import logging -import time from collections.abc import Sequence from enum import Enum, auto from typing import TYPE_CHECKING, Any, NamedTuple, cast @@ -131,9 +130,9 @@ async def relay_agent_response( still working on it. """ offset = min_offset - # perf_counter keeps its resolution on Windows, where monotonic() is coarse. - deadline = time.perf_counter() + timeout - while (remaining := deadline - time.perf_counter()) > 0: + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while (remaining := deadline - loop.time()) > 0: polled = await next_agent_events( app=app, session_id=session_id, @@ -199,6 +198,8 @@ def _message_parts(data: Any) -> tuple[str, list[str]]: async def _back_off(deadline: float) -> None: - backoff = min(EMPTY_READ_BACKOFF_SECONDS, deadline - time.perf_counter()) + backoff = min( + EMPTY_READ_BACKOFF_SECONDS, deadline - asyncio.get_running_loop().time() + ) if backoff > 0: await asyncio.sleep(backoff) diff --git a/tests/adapters/parlant/conftest.py b/tests/adapters/parlant/conftest.py index 772534584..074afe156 100644 --- a/tests/adapters/parlant/conftest.py +++ b/tests/adapters/parlant/conftest.py @@ -6,14 +6,17 @@ from __future__ import annotations +import asyncio import sys -from collections.abc import Awaitable, Callable, Iterator +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator from datetime import UTC, datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest +import pytest_asyncio from band.adapters.parlant import ParlantAdapter, ParlantAdapterConfig +from band.adapters.parlant.messages import process_session_tag from band.core.types import PlatformMessage from band.testing import FakeAgentTools from tests.adapters.parlant.helpers import ( @@ -84,6 +87,26 @@ def background_tasks() -> RecordingBackgroundTasks: return RecordingBackgroundTasks() +@pytest_asyncio.fixture(loop_scope="function") +async def processing_session( + background_tasks: RecordingBackgroundTasks, +) -> AsyncIterator[asyncio.Task[None]]: + """A session whose processing can finish only when cancelled.""" + entered = asyncio.Event() + + async def process() -> None: + entered.set() + await asyncio.Future[None]() + + task = await background_tasks.start(process(), tag=process_session_tag(SESSION_ID)) + try: + await entered.wait() + yield task + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + @pytest.fixture def background_task_service_class() -> Iterator[MagicMock]: """Fake ``parlant.core.background_tasks`` so on_started's import resolves.""" diff --git a/tests/adapters/parlant/helpers.py b/tests/adapters/parlant/helpers.py index dd4f0f69c..883cc2401 100644 --- a/tests/adapters/parlant/helpers.py +++ b/tests/adapters/parlant/helpers.py @@ -2,6 +2,9 @@ from __future__ import annotations +import asyncio +from collections.abc import Coroutine +from typing import Any from unittest.mock import MagicMock from pydantic import BaseModel @@ -46,6 +49,20 @@ class RecordingBackgroundTasks: def __init__(self) -> None: self.cancelled: list[str] = [] + self._tasks: dict[str, asyncio.Task[None]] = {} + + async def start( + self, coroutine: Coroutine[Any, Any, None], /, *, tag: str + ) -> asyncio.Task[None]: + existing = self._tasks.get(tag) + if existing is not None and not existing.done(): + raise RuntimeError(f"Task {tag!r} is already running") + task = asyncio.create_task(coroutine) + self._tasks[tag] = task + return task async def cancel(self, *, tag: str, reason: str = "(not given)") -> None: self.cancelled.append(tag) + task = self._tasks.get(tag) + if task is not None and not task.done(): + task.cancel(reason) diff --git a/tests/adapters/parlant/test_timeout.py b/tests/adapters/parlant/test_timeout.py new file mode 100644 index 000000000..4a86649f5 --- /dev/null +++ b/tests/adapters/parlant/test_timeout.py @@ -0,0 +1,30 @@ +"""Response timeout stops processing before the Band turn returns.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable + +import pytest + +from band.adapters.parlant import ParlantAdapter +from band.adapters.parlant.messages import process_session_tag +from tests.adapters.parlant.helpers import SESSION_ID, RecordingBackgroundTasks + +pytestmark = [pytest.mark.usefixtures("parlant_sessions"), pytest.mark.looptime] + + +class TestParlantTimeout: + async def test_timeout_cancels_and_joins_processing( + self, + start_adapter: Callable[..., Awaitable[ParlantAdapter]], + run_turn: Callable[..., Awaitable[None]], + background_tasks: RecordingBackgroundTasks, + processing_session: asyncio.Task[None], + ) -> None: + adapter = await start_adapter() + + await run_turn(adapter) + + assert background_tasks.cancelled == [process_session_tag(SESSION_ID)] + assert processing_session.done() diff --git a/tests/framework_conformance/stopprobes/__init__.py b/tests/framework_conformance/stopprobes/__init__.py index 1f9b46b8c..450f33f75 100644 --- a/tests/framework_conformance/stopprobes/__init__.py +++ b/tests/framework_conformance/stopprobes/__init__.py @@ -17,6 +17,8 @@ from dataclasses import dataclass from enum import StrEnum +import pytest + from band.client.streaming import ControlMode from band.core.simple_adapter import SimpleAdapter from band.core.turn_lifecycle import BackendTurns @@ -47,6 +49,7 @@ class StopCase: run: Callable[[], Awaitable[list[str]]] expected: list[str] + marks: tuple[pytest.MarkDecorator, ...] = () STOP_WAITS = frozenset({BackendTurns.settled.__code__, BackendTurns.drain.__code__}) diff --git a/tests/framework_conformance/stopprobes/parlant.py b/tests/framework_conformance/stopprobes/parlant.py index 6454e2a2e..1cf08a7d3 100644 --- a/tests/framework_conformance/stopprobes/parlant.py +++ b/tests/framework_conformance/stopprobes/parlant.py @@ -14,9 +14,12 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import Any +from unittest.mock import patch + +import pytest from band.adapters.parlant import ParlantAdapter, ParlantAdapterConfig -from band.adapters.parlant.messages import live_preamble_tasks, process_session_tag +from band.adapters.parlant.messages import process_session_tag from band.client.streaming import ControlMode from band.testing.fake_tools import FakeAgentTools from tests.framework_conformance.stopprobes import ( @@ -24,7 +27,6 @@ StopCase, end_turn, stop_room, - wait_until, ) from tests.framework_conformance.turnprobes import ROOM_ID, turn_input @@ -35,10 +37,15 @@ ENGINE_STOPPED = "engine stopped processing the session" PREAMBLE_STOPPED = "preamble generation stopped" TAG_IS_LIVE = "process_session_tag names the engine's running task" +RESPONSE_TIMEOUT_SECONDS = 5.0 +# Binary ticks keep Parlant's exact-zero expiry check free of float residuals. +VIRTUAL_CLOCK_RESOLUTION_SECONDS = 2**-20 @asynccontextmanager -async def started() -> AsyncIterator[tuple[ParlantAdapter, Any]]: +async def started( + config: ParlantAdapterConfig | None = None, +) -> AsyncIterator[tuple[ParlantAdapter, Any]]: """A started adapter on an owned server whose model never answers.""" import parlant.sdk as p # noqa: PLC0415 -- dev-parlant extra @@ -48,7 +55,7 @@ async def started() -> AsyncIterator[tuple[ParlantAdapter, Any]]: nlp = StubNLP() adapter = ParlantAdapter( - ParlantAdapterConfig(), + config or ParlantAdapterConfig(), nlp_service=nlp.factory, server_options={"host": LOOPBACK, "log_level": p.LogLevel.WARNING}, ) @@ -79,25 +86,92 @@ async def processing_turn( async def end_running_turn(mode: ControlMode) -> list[str]: - async with started() as (adapter, nlp): + async with started() as (adapter, nlp), captured_preambles() as (created, preamble): tools = FakeAgentTools() turn, session_id = await processing_turn(adapter, nlp, tools) engine = background_tasks(adapter)._tasks[process_session_tag(session_id)] - preamble: list[asyncio.Task[Any]] = [] + await created.wait() + await nlp.generation_started(preamble[0]).wait() + await end_turn(adapter, turn, ROOM_ID, mode) + return stopped_work(engine, preamble, tools) - def preamble_is_running() -> bool: - preamble[:] = live_preamble_tasks(session_id) - return bool(preamble) - await wait_until(preamble_is_running) - await end_turn(adapter, turn, ROOM_ID, mode) - stopped, _ = await asyncio.wait({engine}, timeout=ENGINE_WAIT_SECONDS) - preamble_stopped = bool(preamble) and all(task.done() for task in preamble) - return [ - *([ENGINE_STOPPED] if stopped else []), - *([PREAMBLE_STOPPED] if preamble_stopped else []), - *(f"posted {sent}" for sent in tools.messages_sent), - ] +@asynccontextmanager +async def captured_preambles() -> AsyncIterator[ + tuple[asyncio.Event, list[asyncio.Task[Any]]] +]: + """Observe the actual child factory independently of adapter discovery.""" + from parlant.core.engines.alpha.engine import ( # noqa: PLC0415 -- dev-parlant extra + AlphaEngine, + ) + + create_preamble = AlphaEngine._get_preamble_task + children: list[asyncio.Task[Any]] = [] + created = asyncio.Event() + + async def capture(engine: Any, context: Any) -> asyncio.Task[bool]: + child = await create_preamble(engine, context) + children.append(child) + created.set() + return child + + with patch.object(AlphaEngine, "_get_preamble_task", capture): + try: + yield created, children + finally: + for child in children: + child.cancel() + await asyncio.gather(*children, return_exceptions=True) + + +def stopped_work( + engine: asyncio.Task[Any], preamble: list[asyncio.Task[Any]], tools: FakeAgentTools +) -> list[str]: + return [ + *([ENGINE_STOPPED] if engine.done() else []), + *([PREAMBLE_STOPPED] if preamble and all(t.done() for t in preamble) else []), + *(f"posted {sent}" for sent in tools.messages_sent), + ] + + +async def response_timeout() -> list[str]: + config = ParlantAdapterConfig( + response_timeout=RESPONSE_TIMEOUT_SECONDS, + response_poll=RESPONSE_TIMEOUT_SECONDS, + ) + async with ( + started(config) as (adapter, nlp), + captured_preambles() as (created, preamble), + ): + assert adapter._rooms is not None + sessions = adapter._rooms.app.sessions + post = sessions.create_customer_message + processing: list[asyncio.Task[Any]] = [] + + async def post_when_generating(**kwargs: Any) -> Any: + event = await post(**kwargs) + parent = background_tasks(adapter)._tasks[ + process_session_tag(kwargs["session_id"]) + ] + processing.append(parent) + # The response budget starts after this returns to the adapter. + await created.wait() + await nlp.generation_started(preamble[0]).wait() + assert not parent.done() and not preamble[0].done() + return event + + tools = FakeAgentTools() + try: + with patch.object( + sessions, "create_customer_message", post_when_generating + ): + await adapter.on_event(turn_input(tools)) + [engine] = processing + return stopped_work(engine, preamble, tools) + finally: + for parent in processing: + parent.cancel() + await asyncio.gather(*processing, return_exceptions=True) async def tag_parity() -> list[str]: @@ -138,5 +212,10 @@ async def idle() -> list[str]: for mode in ENDING_CONTROLS }, "tag-matches-the-engine": StopCase(run=tag_parity, expected=[TAG_IS_LIVE]), + "response-timeout": StopCase( + run=response_timeout, + expected=[ENGINE_STOPPED, PREAMBLE_STOPPED], + marks=(pytest.mark.looptime(resolution=VIRTUAL_CLOCK_RESOLUTION_SECONDS),), + ), "idle-room": StopCase(run=idle, expected=[]), } diff --git a/tests/framework_conformance/test_backend_stop.py b/tests/framework_conformance/test_backend_stop.py index caa4993ca..948216bb0 100644 --- a/tests/framework_conformance/test_backend_stop.py +++ b/tests/framework_conformance/test_backend_stop.py @@ -68,7 +68,7 @@ } CASES = [ - pytest.param(adapter, case, id=f"{adapter}-{name}") + pytest.param(adapter, case, id=f"{adapter}-{name}", marks=case.marks) for adapter, cases in STOP_PROBES.items() for name, case in cases.items() ] diff --git a/tests/integrations/parlant/stubnlp.py b/tests/integrations/parlant/stubnlp.py index a92f9f349..745650567 100644 --- a/tests/integrations/parlant/stubnlp.py +++ b/tests/integrations/parlant/stubnlp.py @@ -7,7 +7,7 @@ from __future__ import annotations import asyncio -from collections.abc import Mapping +from collections.abc import Callable, Mapping from typing import Any, Generic, TypeVar from parlant.core.nlp.embedding import Embedder, EmbeddingResult @@ -25,10 +25,18 @@ class ParkedGenerator(SchematicGenerator[T], Generic[T]): """Every generation parks, so a turn stays in the engine until cancelled.""" - def __init__(self, entered: asyncio.Event) -> None: + def __init__( + self, + entered: asyncio.Event, + generation_started: Callable[[asyncio.Task[Any]], asyncio.Event], + ) -> None: self._entered = entered + self._generation_started = generation_started async def generate(self, prompt: Any, hints: Mapping[str, Any] = {}) -> Any: + task = asyncio.current_task() + assert task is not None + self._generation_started(task).set() self._entered.set() await asyncio.Future[None]() @@ -73,6 +81,11 @@ class StubNLP(NLPService): def __init__(self) -> None: self.entered = asyncio.Event() + self._generation_started: dict[asyncio.Task[Any], asyncio.Event] = {} + + def generation_started(self, task: asyncio.Task[Any]) -> asyncio.Event: + """Signal when this engine task has entered model generation.""" + return self._generation_started.setdefault(task, asyncio.Event()) def factory(self, container: Any) -> NLPService: """The ``nlp_service=`` callable a Parlant server takes.""" @@ -85,7 +98,7 @@ def supports_streaming(self) -> bool: async def get_schematic_generator( self, t: type[T], hints: Mapping[str, Any] = {} ) -> SchematicGenerator[T]: - return ParkedGenerator[t](self.entered) + return ParkedGenerator[t](self.entered, self.generation_started) async def get_streaming_text_generator(self, hints: Mapping[str, Any] = {}) -> Any: raise NotImplementedError From 77e5a2d2b3d334ff879f431d17c68c57837fa84d Mon Sep 17 00:00:00 2001 From: Alexander Zaikman Date: Sun, 18 Oct 2026 09:29:40 +0300 Subject: [PATCH 3/3] test: make Parlant clocks portable and randomize test order --- pyproject.toml | 1 + .../stopprobes/parlant.py | 63 ++++++++++++------- .../test_parlant_job_coverage.py | 1 + tests/toolkit/clock.py | 15 +++++ tests/toolkit/test_clock.py | 51 +++++++++++++++ uv.lock | 14 +++++ 6 files changed, 121 insertions(+), 24 deletions(-) create mode 100644 tests/toolkit/clock.py create mode 100644 tests/toolkit/test_clock.py diff --git a/pyproject.toml b/pyproject.toml index 4f7f90a6d..efafb42af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -310,6 +310,7 @@ dev-parlant = [ "pydantic-settings>=2.0.0", "pytest>=7.0.0", "pytest-asyncio>=0.21.0", + "pytest-randomly>=5.0.0", "looptime>=0.7", "pytest-mock>=3.10.0", "pytest-cov>=4.0.0", diff --git a/tests/framework_conformance/stopprobes/parlant.py b/tests/framework_conformance/stopprobes/parlant.py index 1cf08a7d3..f7b336515 100644 --- a/tests/framework_conformance/stopprobes/parlant.py +++ b/tests/framework_conformance/stopprobes/parlant.py @@ -11,7 +11,7 @@ from __future__ import annotations import asyncio -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import asynccontextmanager from typing import Any from unittest.mock import patch @@ -29,17 +29,16 @@ stop_room, ) from tests.framework_conformance.turnprobes import ROOM_ID, turn_input +from tests.toolkit.clock import binary_clock_resolution LOOPBACK = "127.0.0.1" -# Real engine work (startup aside) settles well within this, and it stays -# under the suite's 30s test timeout so a turn that never stops fails here. +# Bound virtual waiting when the engine never reaches the model barrier. ENGINE_WAIT_SECONDS = 15.0 ENGINE_STOPPED = "engine stopped processing the session" PREAMBLE_STOPPED = "preamble generation stopped" TAG_IS_LIVE = "process_session_tag names the engine's running task" +# Parlant's exact-zero expiry check needs an integral binary-clock budget. RESPONSE_TIMEOUT_SECONDS = 5.0 -# Binary ticks keep Parlant's exact-zero expiry check free of float residuals. -VIRTUAL_CLOCK_RESOLUTION_SECONDS = 2**-20 @asynccontextmanager @@ -59,8 +58,8 @@ async def started( nlp_service=nlp.factory, server_options={"host": LOOPBACK, "log_level": p.LogLevel.WARNING}, ) - await adapter.on_started("Agent", "A Parlant agent") try: + await adapter.on_started("Agent", "A Parlant agent") yield adapter, nlp finally: await adapter.cleanup_all() @@ -74,26 +73,31 @@ def background_tasks(adapter: ParlantAdapter) -> Any: return adapter.server.container[BackgroundTaskService] +@asynccontextmanager async def processing_turn( adapter: ParlantAdapter, nlp: Any, tools: FakeAgentTools -) -> tuple[asyncio.Task[None], str]: +) -> AsyncIterator[tuple[asyncio.Task[None], str]]: """A turn the engine is generating for, and its session id.""" turn = asyncio.create_task(adapter.on_event(turn_input(tools))) - await asyncio.wait_for(nlp.entered.wait(), ENGINE_WAIT_SECONDS) - assert adapter._rooms is not None - session_id = await adapter._rooms.session_for(ROOM_ID, customer_name="") - return turn, str(session_id) + try: + await asyncio.wait_for(nlp.entered.wait(), ENGINE_WAIT_SECONDS) + assert adapter._rooms is not None + session_id = await adapter._rooms.session_for(ROOM_ID, customer_name="") + yield turn, str(session_id) + finally: + turn.cancel() + await asyncio.gather(turn, return_exceptions=True) async def end_running_turn(mode: ControlMode) -> list[str]: async with started() as (adapter, nlp), captured_preambles() as (created, preamble): tools = FakeAgentTools() - turn, session_id = await processing_turn(adapter, nlp, tools) - engine = background_tasks(adapter)._tasks[process_session_tag(session_id)] - await created.wait() - await nlp.generation_started(preamble[0]).wait() - await end_turn(adapter, turn, ROOM_ID, mode) - return stopped_work(engine, preamble, tools) + async with processing_turn(adapter, nlp, tools) as (turn, session_id): + engine = background_tasks(adapter)._tasks[process_session_tag(session_id)] + await created.wait() + await nlp.generation_started(preamble[0]).wait() + await end_turn(adapter, turn, ROOM_ID, mode) + return stopped_work(engine, preamble, tools) @asynccontextmanager @@ -175,8 +179,10 @@ async def post_when_generating(**kwargs: Any) -> Any: async def tag_parity() -> list[str]: - async with started() as (adapter, nlp): - turn, session_id = await processing_turn(adapter, nlp, FakeAgentTools()) + async with ( + started() as (adapter, nlp), + processing_turn(adapter, nlp, FakeAgentTools()) as (turn, session_id), + ): live = process_session_tag(session_id) in background_tasks(adapter)._tasks await end_turn(adapter, turn, ROOM_ID, ControlMode.STOP) return [TAG_IS_LIVE] if live else [] @@ -203,19 +209,28 @@ async def idle() -> list[str]: return cancelled +def engine_case( + run: Callable[[], Awaitable[list[str]]], expected: list[str] +) -> StopCase: + return StopCase( + run=run, + expected=expected, + marks=(pytest.mark.looptime(resolution=binary_clock_resolution()),), + ) + + CASES: dict[str, StopCase] = { **{ - f"{mode}-mid-turn": StopCase( + f"{mode}-mid-turn": engine_case( run=lambda mode=mode: end_running_turn(mode), expected=[ENGINE_STOPPED, PREAMBLE_STOPPED], ) for mode in ENDING_CONTROLS }, - "tag-matches-the-engine": StopCase(run=tag_parity, expected=[TAG_IS_LIVE]), - "response-timeout": StopCase( + "tag-matches-the-engine": engine_case(run=tag_parity, expected=[TAG_IS_LIVE]), + "response-timeout": engine_case( run=response_timeout, expected=[ENGINE_STOPPED, PREAMBLE_STOPPED], - marks=(pytest.mark.looptime(resolution=VIRTUAL_CLOCK_RESOLUTION_SECONDS),), ), - "idle-room": StopCase(run=idle, expected=[]), + "idle-room": engine_case(run=idle, expected=[]), } diff --git a/tests/framework_conformance/test_parlant_job_coverage.py b/tests/framework_conformance/test_parlant_job_coverage.py index b0bf342e3..7282d73fe 100644 --- a/tests/framework_conformance/test_parlant_job_coverage.py +++ b/tests/framework_conformance/test_parlant_job_coverage.py @@ -17,6 +17,7 @@ # dep fails there rather than silently narrowing what this guard detects. _PARLANT_ONLY_MODULES = { "parlant": "parlant", + "pytest-randomly": "pytest_randomly", "werkzeug": "werkzeug", } diff --git a/tests/toolkit/clock.py b/tests/toolkit/clock.py new file mode 100644 index 000000000..59145cd6b --- /dev/null +++ b/tests/toolkit/clock.py @@ -0,0 +1,15 @@ +"""Virtual clock precision compatible with native asyncio timer wakeups.""" + +from __future__ import annotations + +from math import frexp, ldexp +from time import get_clock_info + + +def binary_clock_resolution() -> float: + """Keep rounded timer remainders within the host's wakeup tolerance. + + Binary ticks also preserve exact integral timeout budgets. + """ + _, exponent = frexp(get_clock_info("monotonic").resolution) + return ldexp(1.0, exponent - 1) diff --git a/tests/toolkit/test_clock.py b/tests/toolkit/test_clock.py new file mode 100644 index 000000000..eaee597fb --- /dev/null +++ b/tests/toolkit/test_clock.py @@ -0,0 +1,51 @@ +"""Virtual timers progress at the precision of each supported host clock.""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import looptime +import pytest + +from tests.toolkit.clock import binary_clock_resolution + + +class TestClock: + @pytest.mark.parametrize( + "native_resolution", [1e-9, 1e-7, 4.166666666666667e-8, 0.015625] + ) + def test_timers_and_integral_deadlines_progress( + self, monkeypatch: pytest.MonkeyPatch, native_resolution: float + ) -> None: + monkeypatch.setattr( + "tests.toolkit.clock.get_clock_info", + lambda name: SimpleNamespace(resolution=native_resolution), + ) + loop = looptime.patch_event_loop( + asyncio.SelectorEventLoop(), resolution=binary_clock_resolution() + ) + # Model asyncio's native wakeup tolerance on every host. + loop._clock_resolution = native_resolution + + async def elapsed_timers() -> list[float]: + elapsed = [] + for delay in (0.05, 0.1, 0.2, 1.0, 1.8377008259506757, 5.0): + start = loop.time() + await asyncio.sleep(delay) + elapsed.append(loop.time() - start) + start = loop.time() + with pytest.raises(TimeoutError): + async with asyncio.timeout(5.0): + await asyncio.Future() + elapsed.append(loop.time() - start) + return elapsed + + try: + assert loop.run_until_complete(elapsed_timers()) == pytest.approx( + [0.05, 0.1, 0.2, 1.0, 1.8377008259506757, 5.0, 5.0], + rel=0, + abs=native_resolution, + ) + finally: + loop.close() diff --git a/uv.lock b/uv.lock index 9a9e11257..37ee8aa3b 100644 --- a/uv.lock +++ b/uv.lock @@ -870,6 +870,7 @@ dev-parlant = [ { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "pytest-mock" }, + { name = "pytest-randomly" }, { name = "pytest-rerunfailures" }, { name = "pytest-timeout" }, { name = "rich", version = "15.0.0", source = { registry = "https://pypi.org/simple" } }, @@ -1120,6 +1121,7 @@ requires-dist = [ { name = "pytest-mock", marker = "extra == 'dev-crewai'", specifier = ">=3.10.0" }, { name = "pytest-mock", marker = "extra == 'dev-parlant'", specifier = ">=3.10.0" }, { name = "pytest-mock", marker = "extra == 'dev-strands-nooa'", specifier = ">=3.10.0" }, + { name = "pytest-randomly", marker = "extra == 'dev-parlant'", specifier = ">=5.0.0" }, { name = "pytest-rerunfailures", marker = "extra == 'dev'", specifier = ">=14.0.0" }, { name = "pytest-rerunfailures", marker = "extra == 'dev-crewai'", specifier = ">=14.0.0" }, { name = "pytest-rerunfailures", marker = "extra == 'dev-parlant'", specifier = ">=14.0.0" }, @@ -7356,6 +7358,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/db/5b/b83a9bf1a3b4ec222f9fa083147ff6816245223da0ab92370e7e056f113f/pytest_mock-3.16.0-py3-none-any.whl", hash = "sha256:007cfeb257801d88d9c0b2a7b5a15a15e73b71968dfd72e7bf8c4a2f8393aec8", size = 10016, upload-time = "2026-09-27T14:57:54.283Z" }, ] +[[package]] +name = "pytest-randomly" +version = "5.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/01/3b/6a40e1b9d925651e601e056a97f60d8a1daeddeac03d5609be60cb4362ce/pytest_randomly-5.0.0.tar.gz", hash = "sha256:e9c575a5873ef168ddbe340ed9e97ce9edb4492ccc821e4b2ac6bb1f0ed515d2", size = 8542, upload-time = "2026-09-01T22:34:20.441Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/10/b4/47e939285caad9a623d021512912ac08dc92a467ad075d179f43729d2934/pytest_randomly-5.0.0-py3-none-any.whl", hash = "sha256:8a0d4703115c0c25b38b6e129fc16b1947b9643ff26a41bc1d185d7e5a7689c1", size = 8920, upload-time = "2026-09-01T22:34:19.227Z" }, +] + [[package]] name = "pytest-rerunfailures" version = "16.7"