feat(olympus): add Telegram routing on live-compatible Hermes baseline

gateway/kanban_watchers.py

@@ -1306,13 +1306,32 @@ def _tick_once_for_board(slug: str) -> "Optional[object]":
                 reconciled = _kb.reconcile_restart_state(
                     conn, olympus_auth=olympus_auth,
                 )
-                if reconciled["effects"] or reconciled["worker_runs"]:
+                terminations = {"executed": 0, "confirmed": 0}
+                telegram_controls = 0
+                if olympus_auth is not None:
+                    terminations = _kb.process_pending_worker_termination_effects(
+                        conn, olympus_auth=olympus_auth,
+                    )
+                    telegram_controls = _kb.reconcile_olympus_telegram_controls(
+                        conn, service_auth=olympus_auth,
+                    )
+                if (
+                    reconciled["effects"]
+                    or reconciled["worker_runs"]
+                    or terminations["executed"]
+                    or terminations["confirmed"]
+                    or telegram_controls
+                ):
                     logger.warning(
                         "kanban dispatcher [%s]: restart reconciliation "
-                        "effects=%d worker_runs=%d",
+                        "effects=%d worker_runs=%d terminations=%d "
+                        "exits=%d telegram_controls=%d",
                         slug,
                         reconciled["effects"],
                         reconciled["worker_runs"],
+                        terminations["executed"],
+                        terminations["confirmed"],
+                        telegram_controls,
                     )
                 return _kb.dispatch_once(
                     conn,

gateway/run.py

@@ -5070,6 +5070,23 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session
             )
             return True  # handled (silently dropped); do not fall through
 
+        # Selected Telegram input is durable Kanban intake, never a
+        # conversational interruption. Persist it before drain/busy handling.
+        routed = await self._route_olympus_telegram_intake(event)
+        if routed is not None:
+            adapter = self.adapters.get(event.source.platform)
+            if adapter:
+                reply_anchor = self._reply_anchor_for_event(event)
+                await adapter._send_with_retry(
+                    chat_id=event.source.chat_id,
+                    content=routed,
+                    reply_to=reply_anchor,
+                    metadata=self._thread_metadata_for_source(
+                        event.source, reply_anchor
+                    ),
+                )
+            return True
+
         # --- Draining case (gateway restarting/stopping) ---
         if self._draining:
             adapter = self._adapter_for_source(event.source)
@@ -8855,6 +8872,12 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
             # clearly moved on.
             _slash_confirm_mod.clear_if_stale(_quick_key)
 
+        # A normal Telegram message in a selected Olympus lane becomes a
+        # durable submission before any running-agent priority/interrupt path.
+        _olympus_routed = await self._route_olympus_telegram_intake(event)
+        if _olympus_routed is not None:
+            return _olympus_routed
+
         # PRIORITY handling when an agent is already running for this session.
         # Default behavior is to interrupt immediately so user text/stop messages
         # are handled with minimal latency.
@@ -9097,6 +9120,13 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
             if _cmd_def_inner and _cmd_def_inner.name == "background":
                 return await self._handle_background_command(event)
 
+            # Olympus is a Telegram control-plane command. Dispatch it while a
+            # conversational agent is active rather than returning the generic
+            # busy response; its own handler enforces platform, authority, and
+            # durable idempotency boundaries.
+            if _cmd_def_inner and _cmd_def_inner.name == "olympus":
+                return await self._handle_olympus_command(event)
+
             # /kanban must bypass the guard. It writes to a profile-agnostic
             # DB (kanban.db), not to the running agent's state. In fact
             # /kanban unblock is often the only way to free a worker that
@@ -9499,6 +9529,9 @@ async def _do_reset():
         if canonical == "kanban":
             return await self._handle_kanban_command(event)
 
+        if canonical == "olympus":
+            return await self._handle_olympus_command(event)
+
         if canonical == "suggestions":
             return await self._handle_suggestions_command(event)
 

gateway/session.py

@@ -12,6 +12,7 @@
 import logging
 import os
 import json
+import re
 import threading
 import uuid
 from pathlib import Path
@@ -56,6 +57,115 @@ def auto_continue_freshness_window() -> float:
         return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT)
 
 
+_OLYMPUS_SELECTION_KEYS = {
+    "schema_version",
+    "board",
+    "root_task_id",
+    "mission_id",
+    "agent_id",
+    "authority_id",
+    "authority_revision",
+    "authority_source",
+    "lease_id",
+    "lease_revision",
+    "lease_source",
+    "scope_digest",
+    "bot_id",
+    "profile",
+    "caller_fingerprint",
+}
+_OLYMPUS_TASK_ID_RE = re.compile(r"^t_[0-9a-f]+$")
+_OLYMPUS_MISSION_ID_RE = re.compile(
+    r"^M-20[0-9]{6}-[a-z0-9][a-z0-9-]{0,39}$"
+)
+_OLYMPUS_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
+_OLYMPUS_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
+
+
+def normalize_olympus_selection(value: Any) -> Dict[str, Any]:
+    """Validate the small durable pointer stored for Telegram routing.
+
+    Authority is deliberately not copied into ``sessions.json``.  The
+    selection points at an existing governed Kanban task; callers must reload
+    and validate that task's current Olympus context before every create or
+    control operation.
+    """
+    if not isinstance(value, dict):
+        raise ValueError("Olympus selection must be an object")
+    unknown = sorted(set(value) - _OLYMPUS_SELECTION_KEYS)
+    if unknown:
+        raise ValueError(
+            "Olympus selection contains unknown field(s): " + ", ".join(unknown)
+        )
+    if value.get("schema_version") != 2:
+        raise ValueError("Olympus selection schema_version must be 2")
+
+    board = str(value.get("board") or "default").strip().lower()
+    root_task_id = str(value.get("root_task_id") or "").strip()
+    mission_id = str(value.get("mission_id") or "").strip()
+    agent_id = str(value.get("agent_id") or "").strip().lower()
+    authority_id = str(value.get("authority_id") or "").strip()
+    authority_source = str(value.get("authority_source") or "").strip()
+    lease_id = str(value.get("lease_id") or "").strip()
+    lease_source = str(value.get("lease_source") or "").strip()
+    bot_id = str(value.get("bot_id") or "").strip()
+    profile = str(value.get("profile") or "").strip().lower()
+    scope_digest = str(value.get("scope_digest") or "").strip().lower()
+    caller_fingerprint = str(
+        value.get("caller_fingerprint") or ""
+    ).strip().lower()
+    if not _OLYMPUS_NAME_RE.fullmatch(board):
+        raise ValueError("Olympus selection board is invalid")
+    if not _OLYMPUS_TASK_ID_RE.fullmatch(root_task_id):
+        raise ValueError("Olympus selection root_task_id is invalid")
+    if not _OLYMPUS_MISSION_ID_RE.fullmatch(mission_id):
+        raise ValueError("Olympus selection mission_id is invalid")
+    if not _OLYMPUS_NAME_RE.fullmatch(agent_id):
+        raise ValueError("Olympus selection agent_id is invalid")
+    if not _OLYMPUS_NAME_RE.fullmatch(profile):
+        raise ValueError("Olympus selection profile is invalid")
+    for field_name, field_value in (
+        ("authority_id", authority_id),
+        ("authority_source", authority_source),
+        ("lease_id", lease_id),
+        ("lease_source", lease_source),
+        ("bot_id", bot_id),
+    ):
+        if not field_value or len(field_value) > 512:
+            raise ValueError(f"Olympus selection {field_name} is invalid")
+    authority_revision = value.get("authority_revision")
+    lease_revision = value.get("lease_revision")
+    if isinstance(authority_revision, bool) or not isinstance(
+        authority_revision, int
+    ) or authority_revision < 1:
+        raise ValueError("Olympus selection authority_revision is invalid")
+    if isinstance(lease_revision, bool) or not isinstance(
+        lease_revision, int
+    ) or lease_revision < 1:
+        raise ValueError("Olympus selection lease_revision is invalid")
+    if not _OLYMPUS_DIGEST_RE.fullmatch(scope_digest):
+        raise ValueError("Olympus selection scope_digest is invalid")
+    if not _OLYMPUS_DIGEST_RE.fullmatch(caller_fingerprint):
+        raise ValueError("Olympus selection caller_fingerprint is invalid")
+    return {
+        "schema_version": 2,
+        "board": board,
+        "root_task_id": root_task_id,
+        "mission_id": mission_id,
+        "agent_id": agent_id,
+        "authority_id": authority_id,
+        "authority_revision": authority_revision,
+        "authority_source": authority_source,
+        "lease_id": lease_id,
+        "lease_revision": lease_revision,
+        "lease_source": lease_source,
+        "scope_digest": scope_digest,
+        "bot_id": bot_id,
+        "profile": profile,
+        "caller_fingerprint": caller_fingerprint,
+    }
+
+
 # ---------------------------------------------------------------------------
 # PII redaction helpers
 # ---------------------------------------------------------------------------
@@ -677,6 +787,10 @@ class SessionEntry:
     # override is rehydrated after a restart and are never written to disk
     # (see sanitize_model_override / SessionStore.set_model_override).
     model_override: Optional[Dict[str, str]] = None
+    # Durable Telegram routing pointer. The referenced Kanban task remains
+    # the source of current authority and lease truth; this is only operator
+    # selection state and never grants permission by itself.
+    olympus_selection: Optional[Dict[str, Any]] = None
 
     def to_dict(self) -> Dict[str, Any]:
         result = {
@@ -713,6 +827,10 @@ def to_dict(self) -> Dict[str, Any]:
             # Defence-in-depth: strip credentials even if a caller stored an
             # unsanitized dict directly on the entry.
             result["model_override"] = sanitize_model_override(self.model_override)
+        if self.olympus_selection is not None:
+            result["olympus_selection"] = normalize_olympus_selection(
+                self.olympus_selection
+            )
         if self.origin:
             result["origin"] = self.origin.to_dict()
         return result
@@ -748,6 +866,17 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionEntry":
                     f"Invalid {_field}: potential directory traversal detected"
                 )
 
+        olympus_selection = None
+        if data.get("olympus_selection") is not None:
+            try:
+                olympus_selection = normalize_olympus_selection(
+                    data["olympus_selection"]
+                )
+            except (TypeError, ValueError) as exc:
+                logger.warning(
+                    "Ignoring invalid persisted Olympus selection: %s", exc
+                )
+
         return cls(
             session_key=session_key,
             session_id=session_id,
@@ -775,6 +904,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionEntry":
             auto_reset_reason=data.get("auto_reset_reason"),
             reset_had_activity=data.get("reset_had_activity", False),
             model_override=sanitize_model_override(data.get("model_override")),
+            olympus_selection=olympus_selection,
         )
 
 
@@ -1470,6 +1600,12 @@ def get_or_create_session(
         with self._lock:
             self._ensure_loaded_locked()
 
+            preserved_olympus_selection = None
+            if session_key in self._entries:
+                preserved_olympus_selection = self._entries[
+                    session_key
+                ].olympus_selection
+
             if session_key in self._entries and not force_new:
                 entry = self._entries[session_key]
                 self._heal_compression_tip_locked(
@@ -1590,6 +1726,7 @@ def get_or_create_session(
                 was_auto_reset=was_auto_reset,
                 auto_reset_reason=auto_reset_reason,
                 reset_had_activity=reset_had_activity,
+                olympus_selection=preserved_olympus_selection,
             )
 
             self._entries[session_key] = entry
@@ -1676,6 +1813,79 @@ def get_model_override(self, session_key: str) -> Optional[Dict[str, str]]:
                 return None
             return dict(entry.model_override) if entry.model_override else None
 
+    def get_olympus_selection(
+        self, session_key: str
+    ) -> Optional[Dict[str, Any]]:
+        """Return a defensive copy of this session's routing selection."""
+        with self._lock:
+            self._ensure_loaded_locked()
+            entry = self._entries.get(session_key)
+            if entry is None or entry.olympus_selection is None:
+                return None
+            return dict(normalize_olympus_selection(entry.olympus_selection))
+
+    def set_olympus_selection(
+        self,
+        session_key: str,
+        selection: Optional[Dict[str, Any]],
+    ) -> bool:
+        """Persist or clear a routing selection on an existing session."""
+        normalized = (
+            normalize_olympus_selection(selection)
+            if selection is not None
+            else None
+        )
+        with self._lock:
+            self._ensure_loaded_locked()
+            entry = self._entries.get(session_key)
+            if entry is None:
+                return False
+            entry.olympus_selection = normalized
+            entry.updated_at = _now()
+            self._save()
+            return True
+
+    def compare_and_set_olympus_selection(
+        self,
+        session_key: str,
+        *,
+        expected: Optional[Dict[str, Any]],
+        replacement: Optional[Dict[str, Any]],
+    ) -> bool:
+        """Replace an Olympus selection only when the captured value matches.
+
+        Verification happens outside the session lock because it may open the
+        Kanban database and call the canonical authority issuer.  This CAS is
+        the persistence boundary that prevents a stale ``/olympus clear`` from
+        deleting a newer selection installed while that verification ran.
+        """
+        normalized_expected = (
+            normalize_olympus_selection(expected)
+            if expected is not None
+            else None
+        )
+        normalized_replacement = (
+            normalize_olympus_selection(replacement)
+            if replacement is not None
+            else None
+        )
+        with self._lock:
+            self._ensure_loaded_locked()
+            entry = self._entries.get(session_key)
+            if entry is None:
+                return False
+            current = (
+                normalize_olympus_selection(entry.olympus_selection)
+                if entry.olympus_selection is not None
+                else None
+            )
+            if current != normalized_expected:
+                return False
+            entry.olympus_selection = normalized_replacement
+            entry.updated_at = _now()
+            self._save()
+            return True
+
     def suspend_session(self, session_key: str) -> bool:
         """Mark a session as suspended so it auto-resets on next access.
 
@@ -1861,6 +2071,7 @@ def reset_session(self, session_key: str, display_name: Optional[str] = None) ->
                 platform=old_entry.platform,
                 chat_type=old_entry.chat_type,
                 is_fresh_reset=True,
+                olympus_selection=old_entry.olympus_selection,
             )
 
             self._entries[session_key] = new_entry
@@ -1930,6 +2141,7 @@ def switch_session(self, session_key: str, target_session_id: str) -> Optional[S
                 display_name=old_entry.display_name,
                 platform=old_entry.platform,
                 chat_type=old_entry.chat_type,
+                olympus_selection=old_entry.olympus_selection,
             )
 
             self._entries[session_key] = new_entry