diff --git a/examples/scenarios/self_start_kickoff.py b/examples/scenarios/self_start_kickoff.py new file mode 100644 index 000000000..e4ffbe66d --- /dev/null +++ b/examples/scenarios/self_start_kickoff.py @@ -0,0 +1,105 @@ +# /// script +# requires-python = ">=3.11" +# dependencies = ["thenvoi-sdk[anthropic]", "python-dotenv"] +# +# [tool.uv.sources] +# thenvoi-sdk = { git = "https://github.com/thenvoi/thenvoi-sdk-python.git" } +# /// +""" +Self-starting agent example using ``Agent.bootstrap_room_message()``. + +Most agents react to messages from the platform. Sometimes you want the +opposite: the agent should start working on its own, with an initial +message that did NOT come from a real user — for example, a webhook that +just received an event, or a cron job that wakes the agent every morning. + +``Agent.bootstrap_room_message(content)`` does exactly that. Pass a +plain string and it creates a fresh chat room, then injects a synthetic +in-memory message so the adapter starts processing as if the user had +typed it. The message is NOT persisted on the platform and other +participants never see it. + +For external systems that need stable ids for retry/replay idempotency, +pass a ``PlatformMessage`` with your own ``id`` and a ``room_id``. + +Run with: + uv run examples/scenarios/self_start_kickoff.py +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import sys + +from dotenv import load_dotenv + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from anthropic.setup_logging import setup_logging # noqa: E402 +from thenvoi import Agent # noqa: E402 +from thenvoi.adapters import AnthropicAdapter # noqa: E402 +from thenvoi.config import load_agent_config # noqa: E402 + +setup_logging() +logger = logging.getLogger(__name__) + + +KICKOFF_MESSAGE = """\ +You are starting in a fresh chat room with no other participants yet. +Your job: find out the current weather in San Francisco and Paris. + +You don't have a weather tool yourself. Use the platform to recruit +help: + +1. Look up peers (agents or users) that can help with weather, web + search, or general lookups. +2. Add the most promising candidate to this room. +3. Ask them, in this room, for the current weather in both cities. +4. When you have answers from the collaborator, summarize which city is + more pleasant right now and stop. + +Prefer asking real collaborators over guessing. If no peer is available, +say so plainly instead of fabricating numbers. +""" + + +async def main() -> None: + load_dotenv() + + ws_url = os.getenv("THENVOI_WS_URL") + rest_url = os.getenv("THENVOI_REST_URL") + if not ws_url: + raise ValueError("THENVOI_WS_URL environment variable is required") + if not rest_url: + raise ValueError("THENVOI_REST_URL environment variable is required") + + agent_id, api_key = load_agent_config("anthropic_agent") + + adapter = AnthropicAdapter( + model="claude-sonnet-4-6", + prompt=( + "You are a coordinator agent on the Thenvoi platform. You can " + "discover peers, add them to rooms, and message them. Prefer " + "delegating specialized work to peers over answering alone." + ), + ) + + agent = Agent.create( + adapter=adapter, + agent_id=agent_id, + api_key=api_key, + ws_url=ws_url, + rest_url=rest_url, + ) + + async with agent: + room_id = await agent.bootstrap_room_message(KICKOFF_MESSAGE) + logger.info("Bootstrapped agent in room %s", room_id) + # Stay running so the agent can iterate with the peers it invites. + await agent.run_forever() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/thenvoi/agent.py b/src/thenvoi/agent.py index 6b1976e1e..d84823198 100644 --- a/src/thenvoi/agent.py +++ b/src/thenvoi/agent.py @@ -3,6 +3,8 @@ from __future__ import annotations import logging +import uuid +from datetime import datetime, timezone from importlib.metadata import PackageNotFoundError from importlib.metadata import version as _get_version from pathlib import Path @@ -17,7 +19,11 @@ ContactEventConfig, ParticipantAddedCallback, ParticipantRemovedCallback, + PlatformMessage, SessionConfig, + SYNTHETIC_KICKOFF_SENDER_ID, + SYNTHETIC_KICKOFF_SENDER_NAME, + SYNTHETIC_SENDER_TYPE, ) from thenvoi.preprocessing.default import DefaultPreprocessor @@ -329,6 +335,106 @@ async def run_forever(self) -> None: """ await self._runtime.run_forever() + async def bootstrap_room_message( + self, + content: str | PlatformMessage, + *, + room_id: str | None = None, + task_id: str | None = None, + message_type: str = "message", + metadata: dict[str, Any] | None = None, + message_id: str | None = None, + ) -> str: + """ + Start the agent on its own with an initial message that did NOT come + through the platform. + + If ``room_id`` is omitted, a fresh chat room is created (linked to + ``task_id`` when provided). The message is delivered to the adapter + exactly like a real chat message but is NOT persisted on the + platform: nothing is written to the database, no + mark_processing / mark_processed lifecycle, and other participants + never see it. + + Returns as soon as the message is enqueued. The adapter processes it + asynchronously on the execution context's loop. + + Two calling shapes: + + - ``bootstrap_room_message("do the thing")`` — pass a plain string. + The message id, sender identity, and timestamp are filled in. + - ``bootstrap_room_message(PlatformMessage(...), room_id=...)`` — + pass a fully-constructed ``PlatformMessage`` when you need + control over the message ``id`` (external systems often want + stable ids like ``"daily-standup:2026-05-06"`` or + ``"webhook:{event_id}"`` for retry / replay idempotency). The + ``room_id`` arg is required in this shape. + + Args: + content: The initial message — either a plain string or a + fully-constructed ``PlatformMessage``. + room_id: Existing room to inject into. Required when ``content`` + is a ``PlatformMessage``. When omitted with a string, a new + room is created via REST. + task_id: Optional task association when creating a new room. + Ignored when ``content`` is a ``PlatformMessage`` or + ``room_id`` is given. + message_type: Platform message type. Defaults to ``"message"``. + Ignored when ``content`` is a ``PlatformMessage``. + metadata: Optional metadata dict (mentions/status are filled in + if missing). Ignored when ``content`` is a + ``PlatformMessage``. + message_id: Stable id for the synthetic message. Defaults to a + generated ``"kickoff:{uuid4}"``. Ignored when ``content`` is + a ``PlatformMessage`` (use ``message.id`` instead). + + Returns: + The room id the message was injected into (newly created when + ``room_id`` was None; otherwise the same id that was passed in). + """ + if not self._started: + raise RuntimeError("Agent not started") + + if isinstance(content, PlatformMessage): + if room_id is None: + raise ValueError( + "room_id is required when content is a PlatformMessage" + ) + await self._runtime.bootstrap_room_message(room_id, content) + return room_id + + if room_id is None: + from thenvoi.client.rest import ChatRoomRequest, DEFAULT_REQUEST_OPTIONS + + response = await self._runtime.link.rest.agent_api_chats.create_agent_chat( + chat=ChatRoomRequest(task_id=task_id), + request_options=DEFAULT_REQUEST_OPTIONS, + ) + if not response.data: + raise RuntimeError( + f"create_agent_chat returned no data for task_id={task_id!r}" + ) + room_id = response.data.id + logger.info("Bootstrap created new room: %s", room_id) + + # Sender identity is intentionally fixed to the synthetic constants — + # ExecutionContext relies on (sender_type=System, sender_id=kickoff) + # to skip platform persistence. ExecutionContext.bootstrap_message + # forces these regardless of what is set here. + message = PlatformMessage( + id=message_id or f"kickoff:{uuid.uuid4()}", + room_id=room_id, + content=content, + sender_id=SYNTHETIC_KICKOFF_SENDER_ID, + sender_type=SYNTHETIC_SENDER_TYPE, + sender_name=SYNTHETIC_KICKOFF_SENDER_NAME, + message_type=message_type, + metadata=metadata or {}, + created_at=datetime.now(timezone.utc), + ) + await self._runtime.bootstrap_room_message(room_id, message) + return room_id + async def _on_execute( self, ctx: "ExecutionContext", diff --git a/src/thenvoi/runtime/execution.py b/src/thenvoi/runtime/execution.py index 7de8de651..9d9c529f2 100644 --- a/src/thenvoi/runtime/execution.py +++ b/src/thenvoi/runtime/execution.py @@ -44,7 +44,9 @@ ParticipantRemovedCallback, SessionConfig, SYNTHETIC_SENDER_TYPE, - SYNTHETIC_CONTACT_EVENTS_SENDER_ID, + SYNTHETIC_KICKOFF_SENDER_ID, + SYNTHETIC_KICKOFF_SENDER_NAME, + SYNTHETIC_SENDER_IDS, ) from .retry_tracker import MessageRetryTracker from ._context_serialization import context_item_to_dict @@ -130,6 +132,23 @@ async def on_event(self, event: PlatformEvent) -> None: """Handle a platform event for this room.""" ... + async def bootstrap_message(self, message: PlatformMessage) -> None: + """Inject a synthetic kickoff message into this execution. + + .. versionchanged:: 0.3.0 + Custom ``Execution`` implementations should add ``bootstrap_message()``. + ``AgentRuntime`` falls back safely for legacy implementations that do + not provide it (raises ``RuntimeError`` at the bootstrap call site), + but typed protocol conformance now includes this method. + + Called by ``Agent.kickoff`` and ``Agent.bootstrap_room_message`` to start + the agent on its own with an initial message that did not come through + the platform. The supplied ``PlatformMessage`` is wrapped as a synthetic + ``MessageEvent`` and delivered to the adapter without platform + persistence, retry tracking, or other-participant visibility. + """ + ... + async def request_resync(self) -> None: """Signal the process loop to re-poll /next immediately. @@ -394,10 +413,19 @@ async def on_event(self, event: PlatformEvent) -> None: logger.debug("Event %s enqueued for room %s", event.type, self.room_id) return - # Track first WebSocket message ID for sync point + # Track first WebSocket message ID for sync point. Skip synthetic + # messages (kickoff, contact-event injections) — they aren't in the DB, + # so using their id as the sync marker would prevent + # _synchronize_with_next from ever finding a matching /next row. if isinstance(event, MessageEvent) and self._first_ws_msg_id is None: msg_id = event.payload.id if event.payload else None - if msg_id: + payload = event.payload + is_synthetic = ( + payload is not None + and payload.sender_type == SYNTHETIC_SENDER_TYPE + and payload.sender_id in SYNTHETIC_SENDER_IDS + ) + if msg_id and not is_synthetic: self._first_ws_msg_id = msg_id logger.debug("Sync point marker set: %s", msg_id) @@ -468,6 +496,60 @@ def mark_participants_sent(self) -> None: """Mark current participants as sent to LLM.""" self._last_participants_sent = self._participants.copy() + async def bootstrap_message(self, message: PlatformMessage) -> None: + """ + Inject a synthetic kickoff message into this execution context. + + The supplied PlatformMessage is wrapped in a MessageEvent and enqueued + through the same path real messages take, but with a synthetic sender + identity (sender_type="System", sender_id="kickoff"). _process_event + recognizes this combination and skips all platform persistence + (mark_processing / mark_processed / mark_failed) and retry tracking — + the message exists only in memory. + + The caller-supplied message.id is preserved as-is so external systems + can use stable ids (e.g. ``"daily-standup:2026-05-06"`` or + ``"webhook:{event_id}"``) to dedupe retries and replays. + + Args: + message: PlatformMessage to inject. id, content, sender_name, + message_type, metadata, and created_at are passed through. + sender_type and sender_id are forced to the synthetic identity. + """ + from thenvoi.client.streaming import MessageCreatedPayload, MessageMetadata + + metadata: dict[str, Any] = dict(message.metadata) if message.metadata else {} + metadata.setdefault("mentions", []) + metadata.setdefault("status", "sent") + + created_at_str = ( + message.created_at.isoformat() + if message.created_at + else datetime.now(timezone.utc).isoformat() + ) + + event = MessageEvent( + room_id=self.room_id, + payload=MessageCreatedPayload( + id=message.id, + content=message.content, + sender_id=SYNTHETIC_KICKOFF_SENDER_ID, + sender_type=SYNTHETIC_SENDER_TYPE, + sender_name=message.sender_name or SYNTHETIC_KICKOFF_SENDER_NAME, + message_type=message.message_type, + metadata=MessageMetadata(**metadata), + chat_room_id=self.room_id, + inserted_at=created_at_str, + updated_at=created_at_str, + ), + ) + await self.on_event(event) + logger.info( + "Bootstrap (kickoff) message %s enqueued for room %s", + message.id, + self.room_id, + ) + def inject_system_message(self, message: str) -> None: """ Queue a system message for injection on next processing. @@ -1161,14 +1243,20 @@ async def _process_event(self, event: PlatformEvent) -> None: logger.debug("Skipping self-message %s", msg_id) return - # Detect synthetic messages (e.g., contact events injected into hub room) - # These don't exist in the database, so skip all tracking and marking + # Detect synthetic messages (e.g., contact events injected into hub + # room, or kickoff/bootstrap injections). These don't exist in the + # database, so skip all tracking and marking. Both sender_type AND + # sender_id must match — never broaden to type alone, since real + # platform "System" messages must keep full mark_* lifecycle. is_synthetic = ( payload.sender_type == SYNTHETIC_SENDER_TYPE - and payload.sender_id == SYNTHETIC_CONTACT_EVENTS_SENDER_ID + and payload.sender_id in SYNTHETIC_SENDER_IDS ) if is_synthetic: - logger.debug("Processing synthetic contact event message") + logger.debug( + "Processing synthetic message (sender_id=%s)", + payload.sender_id, + ) msg_id = None # Clear to skip message marking later # Skip all tracking for synthetic messages - go directly to processing else: diff --git a/src/thenvoi/runtime/platform_runtime.py b/src/thenvoi/runtime/platform_runtime.py index 183f268b4..2678abcc9 100644 --- a/src/thenvoi/runtime/platform_runtime.py +++ b/src/thenvoi/runtime/platform_runtime.py @@ -18,6 +18,7 @@ ContactEventStrategy, ParticipantAddedCallback, ParticipantRemovedCallback, + PlatformMessage, SessionConfig, ) from thenvoi_rest.core.api_error import ApiError @@ -204,6 +205,18 @@ async def run_forever(self) -> None: if self._link: await self._link.run_forever() + async def bootstrap_room_message( + self, room_id: str, message: PlatformMessage + ) -> None: + """Inject a synthetic kickoff message into ``room_id``. + + Forwards to ``AgentRuntime.bootstrap_room_message``. Requires the + runtime to be started; raises ``RuntimeError`` otherwise. + """ + if self._runtime is None: + raise RuntimeError("Runtime not started") + await self._runtime.bootstrap_room_message(room_id, message) + async def _fetch_agent_metadata(self) -> None: """Fetch agent metadata from platform.""" if not self._link: diff --git a/src/thenvoi/runtime/presence.py b/src/thenvoi/runtime/presence.py index c7fcde025..523b5e5d8 100644 --- a/src/thenvoi/runtime/presence.py +++ b/src/thenvoi/runtime/presence.py @@ -197,6 +197,15 @@ async def _handle_room_added(self, event: RoomAddedEvent) -> None: logger.warning("room_added event without room_id or payload") return + # Skip if already tracked. Happens when bootstrap_room_message + # (kickoff) claimed the room and subscribed before the platform's + # room_added event arrived. Re-running subscribe_room would issue a + # duplicate Phoenix join. The filter and on_room_joined work was + # already done by the bootstrap path or the original add. + if room_id in self.rooms: + logger.debug("room_added for already-tracked room %s, skipping", room_id) + return + payload = event.payload.model_dump(exclude_none=True) # Apply filter if configured diff --git a/src/thenvoi/runtime/runtime.py b/src/thenvoi/runtime/runtime.py index 618a447b4..11139044b 100644 --- a/src/thenvoi/runtime/runtime.py +++ b/src/thenvoi/runtime/runtime.py @@ -17,6 +17,7 @@ from .types import ( ParticipantAddedCallback, ParticipantRemovedCallback, + PlatformMessage, SessionConfig, ) @@ -245,6 +246,71 @@ async def _on_reconnected(self) -> None: # --- Execution management --- + async def bootstrap_room_message( + self, room_id: str, message: PlatformMessage + ) -> None: + """ + Kick the agent off in ``room_id`` with an initial message that did NOT + come from the platform. + + Use this to start an agent on its own. Common examples: a webhook + handler that creates a fresh chat room and wants the agent to begin + work immediately with full context, a scheduled cron job that wakes + an agent at 9am to file a daily report, or a self-starting demo + agent. The message is delivered to the adapter exactly like a real + chat message, but lives only in memory: nothing is written to the + platform, no mark_processing / mark_processed lifecycle, no retry + tracking, no other participants ever see it. + + The caller's ``message.id`` is preserved so external systems can use + stable ids (e.g. ``"daily-standup:2026-05-06"`` or + ``"webhook:{event_id}"``) for idempotency across retries. + + If the agent has not yet joined ``room_id``, this also subscribes to + the room and creates its execution context, so future real messages + flow through the normal presence pipeline afterwards. + """ + claimed_room = False + created_execution = False + try: + if room_id not in self.presence.rooms: + # Claim the slot before awaiting subscribe_room so concurrent + # bootstrap calls for the same room don't double-subscribe. + self.presence.rooms.add(room_id) + claimed_room = True + try: + await self.link.subscribe_room(room_id) + except Exception: + self.presence.rooms.discard(room_id) + raise + logger.debug( + "Bootstrap subscribed to room %s outside presence flow", room_id + ) + + execution = self.executions.get(room_id) + if execution is None: + execution = await self._create_execution(room_id) + created_execution = True + + bootstrap = getattr(execution, "bootstrap_message", None) + if bootstrap is None: + raise RuntimeError( + f"Execution for room {room_id} does not support bootstrap_message" + ) + await bootstrap(message) + except Exception: + if created_execution: + await self._destroy_execution(room_id) + if claimed_room: + self.presence.rooms.discard(room_id) + try: + await self.link.unsubscribe_room(room_id) + except Exception: + logger.exception( + "Failed to unsubscribe room %s after bootstrap failure", room_id + ) + raise + async def _create_execution(self, room_id: str) -> Execution: """Create and start execution context for a room.""" if room_id in self.executions: @@ -276,7 +342,11 @@ async def _create_execution(self, room_id: str) -> Execution: ) self.executions[room_id] = execution - await execution.start() + try: + await execution.start() + except Exception: + self.executions.pop(room_id, None) + raise logger.debug("Created execution for room %s", room_id) return execution diff --git a/src/thenvoi/runtime/types.py b/src/thenvoi/runtime/types.py index 2e8e3ba60..12381c1ef 100644 --- a/src/thenvoi/runtime/types.py +++ b/src/thenvoi/runtime/types.py @@ -37,6 +37,24 @@ # messages. Used in MessageEvent.payload.sender_name. SYNTHETIC_CONTACT_EVENTS_SENDER_NAME = "Contact Events" +# Sender ID for synthetic kickoff messages — bootstrap_room_message / kickoff() +# inject these to start an agent in a room with an initial message that did not +# come from the platform. Treated like contact-event synthetics: ExecutionContext +# skips mark_processing/mark_processed/retry tracking for them. +SYNTHETIC_KICKOFF_SENDER_ID = "kickoff" +SYNTHETIC_KICKOFF_SENDER_NAME = "Kickoff" + +# Closed set of recognized synthetic sender ids. ExecutionContext requires BOTH +# sender_type == SYNTHETIC_SENDER_TYPE AND sender_id in this set before +# treating a message as synthetic — never broaden to type alone, since real +# platform "System" messages must keep full mark_* lifecycle. +SYNTHETIC_SENDER_IDS: frozenset[str] = frozenset( + { + SYNTHETIC_CONTACT_EVENTS_SENDER_ID, + SYNTHETIC_KICKOFF_SENDER_ID, + } +) + def normalize_handle(handle: str | None) -> str | None: """ diff --git a/tests/runtime/test_kickoff.py b/tests/runtime/test_kickoff.py new file mode 100644 index 000000000..4be9e4316 --- /dev/null +++ b/tests/runtime/test_kickoff.py @@ -0,0 +1,425 @@ +"""Tests for the kickoff / bootstrap_room_message feature. + +These tests cover the synthetic-injection path that lets an agent start work +in a room from an initial message that did not come through the platform. +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from thenvoi.client.streaming import MessageCreatedPayload, MessageMetadata +from thenvoi.platform.event import MessageEvent +from thenvoi.runtime.execution import ExecutionContext +from thenvoi.runtime.types import ( + PlatformMessage, + SYNTHETIC_KICKOFF_SENDER_ID, + SYNTHETIC_SENDER_TYPE, +) + + +@pytest.fixture +def mock_link(): + link = MagicMock() + link.agent_id = "agent-123" + link.rest = MagicMock() + link.rest.agent_api_participants = MagicMock() + link.rest.agent_api_participants.list_agent_chat_participants = AsyncMock( + return_value=MagicMock(data=[]) + ) + link.rest.agent_api_context = MagicMock() + link.rest.agent_api_context.get_agent_chat_context = AsyncMock( + return_value=MagicMock(data=[]) + ) + link.mark_processing = AsyncMock() + link.mark_processed = AsyncMock() + link.mark_failed = AsyncMock() + link.get_next_message = AsyncMock(return_value=None) + link.get_stale_processing_messages = AsyncMock(return_value=[]) + link.subscribe_room = AsyncMock() + link.unsubscribe_room = AsyncMock() + return link + + +def _platform_message( + *, + msg_id: str = "kickoff:test-1", + content: str = "begin work on the ticket", + metadata: dict | None = None, +) -> PlatformMessage: + return PlatformMessage( + id=msg_id, + room_id="room-1", + content=content, + sender_id=SYNTHETIC_KICKOFF_SENDER_ID, + sender_type=SYNTHETIC_SENDER_TYPE, + sender_name="Kickoff", + message_type="message", + metadata=metadata or {}, + created_at=datetime.now(timezone.utc), + ) + + +class TestExecutionBootstrapMessage: + """Direct tests on ExecutionContext.bootstrap_message.""" + + @pytest.mark.asyncio + async def test_delivers_synthetic_event_to_handler(self, mock_link): + handler = AsyncMock() + ctx = ExecutionContext("room-1", mock_link, handler, agent_id="agent-123") + await ctx.start() + try: + await ctx.bootstrap_message(_platform_message(content="hello")) + await asyncio.wait_for(_wait_for_handler(handler), timeout=2.0) + finally: + await ctx.stop() + + # Handler called with a MessageEvent carrying synthetic identity + assert handler.await_count == 1 + delivered_event = handler.await_args.args[1] + assert isinstance(delivered_event, MessageEvent) + assert delivered_event.payload is not None + assert delivered_event.payload.content == "hello" + assert delivered_event.payload.sender_id == SYNTHETIC_KICKOFF_SENDER_ID + assert delivered_event.payload.sender_type == SYNTHETIC_SENDER_TYPE + + @pytest.mark.asyncio + async def test_no_platform_persistence_calls(self, mock_link): + """Synthetic kickoff messages must skip mark_processing/mark_processed.""" + handler = AsyncMock() + ctx = ExecutionContext("room-1", mock_link, handler, agent_id="agent-123") + await ctx.start() + try: + await ctx.bootstrap_message(_platform_message()) + await asyncio.wait_for(_wait_for_handler(handler), timeout=2.0) + finally: + await ctx.stop() + + mock_link.mark_processing.assert_not_called() + mock_link.mark_processed.assert_not_called() + mock_link.mark_failed.assert_not_called() + + @pytest.mark.asyncio + async def test_preserves_caller_message_id(self, mock_link): + """External systems depend on stable ids for retry/replay idempotency.""" + handler = AsyncMock() + ctx = ExecutionContext("room-1", mock_link, handler, agent_id="agent-123") + await ctx.start() + try: + await ctx.bootstrap_message(_platform_message(msg_id="webhook:event-abc")) + await asyncio.wait_for(_wait_for_handler(handler), timeout=2.0) + finally: + await ctx.stop() + + delivered_event = handler.await_args.args[1] + assert delivered_event.payload.id == "webhook:event-abc" + + @pytest.mark.asyncio + async def test_bootstrap_does_not_poison_sync_marker(self, mock_link): + """A kickoff arriving before any real WS message must not become the sync point.""" + handler = AsyncMock() + ctx = ExecutionContext("room-1", mock_link, handler, agent_id="agent-123") + + # Inject a kickoff first + await ctx.bootstrap_message(_platform_message(msg_id="kickoff:should-not-mark")) + assert ctx._first_ws_msg_id is None + + # Then a real WS message + real_event = MessageEvent( + room_id="room-1", + payload=MessageCreatedPayload( + id="real-msg-42", + content="hi", + sender_id="user-1", + sender_type="User", + sender_name="Alice", + message_type="message", + metadata=MessageMetadata(mentions=[], status="sent"), + chat_room_id="room-1", + inserted_at="2024-01-01T00:00:00Z", + updated_at="2024-01-01T00:00:00Z", + ), + ) + await ctx.on_event(real_event) + assert ctx._first_ws_msg_id == "real-msg-42" + + +class TestRealSystemMessageStillTracked: + """Regression: real platform 'System' messages must keep mark_* lifecycle.""" + + @pytest.mark.asyncio + async def test_real_system_message_calls_mark_processed(self, mock_link): + handler = AsyncMock() + ctx = ExecutionContext("room-1", mock_link, handler, agent_id="agent-123") + await ctx.start() + try: + real_system_event = MessageEvent( + room_id="room-1", + payload=MessageCreatedPayload( + id="real-sys-1", + content="moderator note", + sender_id="some-real-system-sender", # NOT a synthetic id + sender_type=SYNTHETIC_SENDER_TYPE, + sender_name="System", + message_type="message", + metadata=MessageMetadata(mentions=[], status="sent"), + chat_room_id="room-1", + inserted_at="2024-01-01T00:00:00Z", + updated_at="2024-01-01T00:00:00Z", + ), + ) + await ctx.on_event(real_system_event) + await asyncio.wait_for(_wait_for_handler(handler), timeout=2.0) + finally: + await ctx.stop() + + mock_link.mark_processing.assert_called_once_with("room-1", "real-sys-1") + mock_link.mark_processed.assert_called_once_with("room-1", "real-sys-1") + + +class TestAgentRuntimeBootstrap: + """AgentRuntime.bootstrap_room_message subscribes + ensures execution.""" + + @pytest.mark.asyncio + async def test_subscribes_when_room_unknown(self, mock_link): + from thenvoi.runtime.runtime import AgentRuntime + + handler = AsyncMock() + runtime = AgentRuntime(mock_link, "agent-123", on_execute=handler) + + await runtime.bootstrap_room_message("new-room", _platform_message()) + + mock_link.subscribe_room.assert_awaited_once_with("new-room") + assert "new-room" in runtime.presence.rooms + assert "new-room" in runtime.executions + + # Cleanup + for room_id in list(runtime.executions.keys()): + await runtime.executions[room_id].stop() + + @pytest.mark.asyncio + async def test_skips_subscribe_when_room_known(self, mock_link): + from thenvoi.runtime.runtime import AgentRuntime + + handler = AsyncMock() + runtime = AgentRuntime(mock_link, "agent-123", on_execute=handler) + runtime.presence.rooms.add("known-room") + + await runtime.bootstrap_room_message("known-room", _platform_message()) + + mock_link.subscribe_room.assert_not_called() + + for room_id in list(runtime.executions.keys()): + await runtime.executions[room_id].stop() + + @pytest.mark.asyncio + async def test_rolls_back_claimed_room_when_execution_lacks_bootstrap( + self, mock_link + ): + from thenvoi.runtime.runtime import AgentRuntime + + class NoBootstrapExecution: + async def start(self): + return None + + async def stop(self, timeout=None): + return True + + runtime = AgentRuntime( + mock_link, + "agent-123", + on_execute=AsyncMock(), + execution_factory=lambda *_args, **_kwargs: NoBootstrapExecution(), + ) + + with pytest.raises(RuntimeError, match="does not support bootstrap_message"): + await runtime.bootstrap_room_message("new-room", _platform_message()) + + assert "new-room" not in runtime.presence.rooms + assert "new-room" not in runtime.executions + mock_link.subscribe_room.assert_awaited_once_with("new-room") + mock_link.unsubscribe_room.assert_awaited_once_with("new-room") + + @pytest.mark.asyncio + async def test_rolls_back_execution_slot_when_start_fails(self, mock_link): + from thenvoi.runtime.runtime import AgentRuntime + + class FailingStartExecution: + async def start(self): + raise RuntimeError("start failed") + + async def stop(self, timeout=None): + return True + + runtime = AgentRuntime( + mock_link, + "agent-123", + on_execute=AsyncMock(), + execution_factory=lambda *_args, **_kwargs: FailingStartExecution(), + ) + + with pytest.raises(RuntimeError, match="start failed"): + await runtime.bootstrap_room_message("new-room", _platform_message()) + + assert "new-room" not in runtime.presence.rooms + assert "new-room" not in runtime.executions + mock_link.subscribe_room.assert_awaited_once_with("new-room") + mock_link.unsubscribe_room.assert_awaited_once_with("new-room") + + +class TestAgentBootstrap: + """Agent.bootstrap_room_message public-entry tests (string + PlatformMessage).""" + + @pytest.fixture + def started_agent(self, mock_link): + """Build a started Agent backed by a real AgentRuntime + mock link. + + Avoids reaching into private attributes — uses the public Agent + constructor with a stubbed PlatformRuntime, then flips the started + flag the same way Agent.start() would. + """ + from thenvoi.agent import Agent + from thenvoi.runtime.runtime import AgentRuntime + + handler = AsyncMock() + agent_runtime = AgentRuntime(mock_link, "agent-123", on_execute=handler) + + platform_runtime = MagicMock() + platform_runtime.link = mock_link + platform_runtime.bootstrap_room_message = AsyncMock( + side_effect=agent_runtime.bootstrap_room_message + ) + + adapter = MagicMock() + adapter.on_started = AsyncMock() + adapter.on_event = AsyncMock() + adapter.on_cleanup = AsyncMock() + + agent = Agent(runtime=platform_runtime, adapter=adapter) + agent._started = True + + return agent, agent_runtime, handler + + @pytest.mark.asyncio + async def test_string_creates_room_and_delivers_to_adapter( + self, started_agent, mock_link + ): + agent, agent_runtime, handler = started_agent + + created = MagicMock() + created.id = "fresh-room-9" + mock_link.rest.agent_api_chats = MagicMock() + mock_link.rest.agent_api_chats.create_agent_chat = AsyncMock( + return_value=MagicMock(data=created) + ) + + room_id = await agent.bootstrap_room_message("go!", task_id="task-7") + + assert room_id == "fresh-room-9" + mock_link.rest.agent_api_chats.create_agent_chat.assert_awaited_once() + assert "fresh-room-9" in agent_runtime.executions + + await asyncio.wait_for(_wait_for_handler(handler), timeout=2.0) + delivered = handler.await_args.args[1] + assert delivered.payload.content == "go!" + assert delivered.payload.sender_id == SYNTHETIC_KICKOFF_SENDER_ID + + for r in list(agent_runtime.executions.keys()): + await agent_runtime.executions[r].stop() + + @pytest.mark.asyncio + async def test_string_raises_when_create_chat_returns_no_data( + self, started_agent, mock_link + ): + agent, _agent_runtime, _handler = started_agent + mock_link.rest.agent_api_chats = MagicMock() + mock_link.rest.agent_api_chats.create_agent_chat = AsyncMock( + return_value=MagicMock(data=None) + ) + with pytest.raises(RuntimeError, match="task_id='task-9'"): + await agent.bootstrap_room_message("go!", task_id="task-9") + + @pytest.mark.asyncio + async def test_platform_message_preserves_caller_id_end_to_end( + self, started_agent, mock_link + ): + """Caller-supplied PlatformMessage id must reach the adapter unchanged.""" + agent, agent_runtime, handler = started_agent + + message = _platform_message(msg_id="webhook:event-xyz", content="run report") + await agent.bootstrap_room_message(message, room_id="room-42") + + await asyncio.wait_for(_wait_for_handler(handler), timeout=2.0) + delivered = handler.await_args.args[1] + assert delivered.payload.id == "webhook:event-xyz" + assert delivered.payload.content == "run report" + + for r in list(agent_runtime.executions.keys()): + await agent_runtime.executions[r].stop() + + @pytest.mark.asyncio + async def test_platform_message_requires_room_id(self, started_agent): + agent, _agent_runtime, _handler = started_agent + with pytest.raises(ValueError, match="room_id is required"): + await agent.bootstrap_room_message(_platform_message()) + + @pytest.mark.asyncio + async def test_raises_when_agent_not_started(self): + from thenvoi.agent import Agent + + adapter = MagicMock() + platform_runtime = MagicMock() + agent = Agent(runtime=platform_runtime, adapter=adapter) + # _started defaults to False + + with pytest.raises(RuntimeError, match="Agent not started"): + await agent.bootstrap_room_message("nope") + + with pytest.raises(RuntimeError, match="Agent not started"): + await agent.bootstrap_room_message(_platform_message(), room_id="room-1") + + +class TestRoomAddedDedupe: + """Regression: WS room_added for a kickoff-claimed room must not double-subscribe.""" + + @pytest.mark.asyncio + async def test_room_added_skips_subscribe_when_room_already_tracked( + self, mock_link + ): + from thenvoi.client.streaming import RoomAddedPayload + from thenvoi.platform.event import RoomAddedEvent + from thenvoi.runtime.presence import RoomPresence + + presence = RoomPresence(mock_link) + # Simulate kickoff having already claimed and subscribed to the room. + presence.rooms.add("claimed-room") + on_joined = AsyncMock() + presence.on_room_joined = on_joined + + event = RoomAddedEvent( + room_id="claimed-room", + payload=RoomAddedPayload( + id="claimed-room", + inserted_at="2024-01-01T00:00:00Z", + updated_at="2024-01-01T00:00:00Z", + ), + ) + await presence._handle_room_added(event) + + # subscribe_room must not be called a second time, and on_room_joined + # should not re-fire (execution would already exist anyway, but the + # callback contract is "once per join"). + mock_link.subscribe_room.assert_not_called() + on_joined.assert_not_called() + + +# --- helpers --- + + +async def _wait_for_handler(handler: AsyncMock) -> None: + while handler.await_count == 0: + await asyncio.sleep(0.01) diff --git a/uv.lock b/uv.lock index 17bb76892..9240152da 100644 --- a/uv.lock +++ b/uv.lock @@ -8152,7 +8152,7 @@ wheels = [ [[package]] name = "thenvoi-sdk" -version = "0.2.8" +version = "0.2.10" source = { editable = "." } dependencies = [ { name = "cryptography" },