feat(agent): add bootstrap_room_message for self-starting agents

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())

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",

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: