feat(memory): add Piri round-trip pipeline

bridge/__main__.py

@@ -2,7 +2,7 @@
 import logging
 import os
 import time
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -51,6 +51,72 @@ class AppContext:
     clock: Any
 
 
+def _build_piri_runtime(settings: Settings) -> Any:
+    """Compose the Piri adapter and its fail-closed audience memory route."""
+
+    from telegram_bot.core.memory_audience import (
+        MemoryAudience,
+        audience_from_piri_environment,
+        shared_memory_audience,
+    )
+    from telegram_bot.core.piri_runtime import PiriRuntime
+    from telegram_bot.memory.distill_types import validate_memory_route
+
+    route_environment_factory: (
+        Callable[[str, str], Mapping[str, str]] | None
+    ) = None
+    memory_environment_validator: (
+        Callable[[Mapping[str, str]], object] | None
+    ) = None
+    if settings.bridge_memory_mode == "audience-scoped":
+        shared = shared_memory_audience(settings)
+
+        def build_piri_route_environment(audience: str, scope: str):
+            validate_memory_route(audience, scope)
+            return MemoryAudience(audience, scope, shared.root).piri_environment(
+                settings
+            )
+
+        def validate_piri_memory_environment(environment: Mapping[str, str]):
+            return audience_from_piri_environment(settings, environment)
+
+        route_environment_factory = build_piri_route_environment
+        memory_environment_validator = validate_piri_memory_environment
+
+    logger.info("Piri provider routed through unrestricted PiriRuntime RPC adapter")
+    return PiriRuntime(
+        executable=settings.piri_cli_path,
+        process_environment=os.environ,
+        model_catalog_directory=str(Path(settings.project_root).resolve()),
+        memory_materializer_path=settings.codex_memory_materializer_path,
+        memory_bootstrap_timeout_seconds=(
+            settings.codex_memory_bootstrap_timeout_seconds
+        ),
+        memory_environment_validator=memory_environment_validator,
+        route_environment_factory=route_environment_factory,
+    )
+
+
+def _build_distill_environment(settings: Settings) -> dict[str, str] | None:
+    """Return the private Codex extraction environment when one is required."""
+
+    if not (
+        settings.agent_provider == "codex"
+        and settings.bridge_memory_mode == "audience-scoped"
+    ):
+        return None
+    from telegram_bot.core.memory_audience import shared_memory_audience
+    from telegram_bot.utils.secure_fs import ensure_private_directory
+
+    environment = dict(os.environ)
+    environment.update(
+        shared_memory_audience(settings).codex_environment(settings)
+    )
+    ensure_private_directory(Path(environment["CODEX_HOME"]))
+    ensure_private_directory(Path(environment["CODEX_SQLITE_HOME"]))
+    return environment
+
+
 def build_context(
     settings: Settings,
     *,
@@ -157,14 +223,7 @@ def route_environment(audience: str, scope: str):
             / claude_project_dir_name(Path(settings.project_root).resolve()),
         )
     elif settings.agent_provider == "piri" and agent_runtime is None:
-        from telegram_bot.core.piri_runtime import PiriRuntime
-
-        logger.info("Piri provider routed through unrestricted PiriRuntime RPC adapter")
-        agent_runtime = PiriRuntime(
-            executable=settings.piri_cli_path,
-            process_environment=os.environ,
-            model_catalog_directory=str(Path(settings.project_root).resolve()),
-        )
+        agent_runtime = _build_piri_runtime(settings)
     telegram_port = telegram_port or Application.builder
     clock = clock or time
     bind_logs_dir(settings.logs_dir)
@@ -194,21 +253,7 @@ def route_environment(audience: str, scope: str):
     honcho_enabled = (
         settings.honcho_memory_enabled
     )
-    distill_environment = None
-    if (
-        settings.agent_provider == "codex"
-        and settings.bridge_memory_mode == "audience-scoped"
-    ):
-        from telegram_bot.core.memory_audience import shared_memory_audience
-
-        distill_environment = dict(os.environ)
-        distill_environment.update(
-            shared_memory_audience(settings).codex_environment(settings)
-        )
-        from telegram_bot.utils.secure_fs import ensure_private_directory
-
-        ensure_private_directory(Path(distill_environment["CODEX_HOME"]))
-        ensure_private_directory(Path(distill_environment["CODEX_SQLITE_HOME"]))
+    distill_environment = _build_distill_environment(settings)
     distill_extraction_worker = project_chat.build_distill_extraction_worker(
         distill_journal,
         CodexExecDistillBackend(
@@ -222,7 +267,7 @@ def route_environment(audience: str, scope: str):
         model=settings.codex_distill_model,
     )
     distill_snapshot_worker = None
-    if settings.agent_provider == "codex":
+    if settings.agent_provider in {"codex", "piri"}:
         from telegram_bot.memory.codex_snapshot import CodexThreadSnapshotter
 
         distill_snapshot_worker = CodexThreadSnapshotter(
@@ -251,7 +296,7 @@ def route_environment(audience: str, scope: str):
             shared_memory_audience(settings).root,
         )
     distill_wiki_sink_worker = None
-    if settings.agent_provider == "codex" and wiki_enabled:
+    if settings.agent_provider in {"codex", "piri"} and wiki_enabled:
         from telegram_bot.memory.distill_wiki_worker import (
             CodexDistillWikiSinkWorker,
         )
@@ -262,7 +307,7 @@ def route_environment(audience: str, scope: str):
             require_memory_route=audience_scoped,
         )
     distill_honcho_sink_worker = None
-    if settings.agent_provider == "codex" and honcho_enabled:
+    if settings.agent_provider in {"codex", "piri"} and honcho_enabled:
         from telegram_bot.memory.distill_honcho_worker import (
             CodexDistillHonchoSinkWorker,
             HonchoHttpSender,

bridge/core/bot.py

@@ -1196,7 +1196,7 @@ async def _enqueue_previous_codex_session(
     ) -> DistillJob | None:
         provider = str(session.get("provider", "claude")).strip().lower()
         thread_id = session.get("session_id")
-        if provider != "codex" or not isinstance(thread_id, str) or not thread_id:
+        if provider not in {"codex", "piri"} or not isinstance(thread_id, str) or not thread_id:
             return None
         journal = getattr(self, "_distill_journal", None)
         if journal is None:
@@ -1221,7 +1221,7 @@ async def _enqueue_previous_codex_session(
                 memory_audience = stored_audience
                 memory_scope = stored_scope
         enqueue_kwargs = {
-            "provider": "codex",
+            "provider": provider,
             "thread_id": thread_id,
             "trigger": trigger,
             "memory_audience": memory_audience,
@@ -1242,7 +1242,7 @@ async def _align_active_provider(
         user_id: int | None = None,
         chat_id: int | None = None,
     ):
-        """Durably capture a departing Codex thread before provider state resets."""
+        """Durably capture a departing writeback-capable thread before reset."""
         if session is None:
             session = await self._session_manager.get_session(session_key)
         provider = str(session.get("provider", "claude")).strip().lower()
@@ -1411,7 +1411,8 @@ async def _save_session_id(
                 raise
             except Exception as error:
                 logger.warning(
-                    "Codex checkpoint accounting failed error=%s",
+                    "%s checkpoint accounting failed error=%s",
+                    self._active_provider().title(),
                     type(error).__name__,
                 )
 
@@ -1471,7 +1472,8 @@ async def _record_codex_checkpoint(
         chat_id: int | None,
     ) -> None:
         """Count completed turns and durably enqueue the first reached gate."""
-        if self._active_provider() != "codex":
+        active_provider = self._active_provider()
+        if active_provider not in {"codex", "piri"}:
             return
         if getattr(self, "_distill_journal", None) is None:
             return
@@ -1530,7 +1532,7 @@ async def _record_codex_checkpoint(
             try:
                 session = await self._session_manager.get_session(session_key)
                 if (
-                    session.get("provider") != "codex"
+                    session.get("provider") != active_provider
                     or session.get("session_id") != thread_id
                 ):
                     progress_by_key.pop(session_key, None)
@@ -1546,7 +1548,8 @@ async def _record_codex_checkpoint(
                 raise
             except Exception as error:
                 logger.warning(
-                    "Codex checkpoint journal enqueue failed error=%s",
+                    "%s checkpoint journal enqueue failed error=%s",
+                    active_provider.title(),
                     type(error).__name__,
                 )
                 return
@@ -1584,7 +1587,7 @@ async def _enqueue_shutdown_distills(
         selected_keys = active_keys[:limit]
         if len(active_keys) > limit:
             logger.warning(
-                "Codex shutdown distill queue capped at %d active sessions",
+                "Memory shutdown distill queue capped at %d active sessions",
                 limit,
             )
 
@@ -1601,7 +1604,7 @@ async def enqueue_selected() -> None:
                     raise
                 except Exception as error:
                     logger.warning(
-                        "Codex shutdown distill queue entry failed error=%s",
+                        "Memory shutdown distill queue entry failed error=%s",
                         type(error).__name__,
                     )
 
@@ -1614,7 +1617,7 @@ async def enqueue_selected() -> None:
             await asyncio.wait_for(enqueue_selected(), timeout=timeout)
         except asyncio.TimeoutError:
             logger.warning(
-                "Codex shutdown distill queue timed out after %.2fs",
+                "Memory shutdown distill queue timed out after %.2fs",
                 timeout,
             )
 

bridge/core/bot_commands.py

@@ -583,7 +583,7 @@ def _explicit_distill_discriminator(session: dict) -> str:
     async def _cmd_distill(
         self, update: Update, context: ContextTypes.DEFAULT_TYPE
     ) -> None:
-        """Durably request write-back for the current Codex thread without reset."""
+        """Durably request write-back for the current supported thread without reset."""
 
         del context
         if not await self._check_access(update):
@@ -593,11 +593,13 @@ async def _cmd_distill(
         chat = self._require_chat(update)
         log_debug(user_id, "command", "/distill")
 
-        if self._active_provider() != "codex":
-            reply = "ℹ️ /distill is available only for active Codex sessions."
+        active_provider = self._active_provider()
+        if active_provider not in {"codex", "piri"}:
+            reply = "ℹ️ /distill is available only for active Codex or Piri sessions."
             await message.reply_text(reply)
             log_debug(user_id, "bot", reply)
             return
+        provider_label = active_provider.title()
 
         conversation_key = self._conversation_key(user_id, chat.id)
         tasks = getattr(self, "_tasks", None)
@@ -608,7 +610,7 @@ async def _cmd_distill(
         )
         if active_task is not None and not active_task.done():
             reply = (
-                "⏳ The current Codex turn is still running. "
+                f"⏳ The current {provider_label} turn is still running. "
                 "Run /distill again after it finishes."
             )
             await message.reply_text(reply)
@@ -617,11 +619,11 @@ async def _cmd_distill(
         session = await self._session_manager.get_session(conversation_key)
         thread_id = session.get("session_id")
         if (
-            str(session.get("provider", "")).strip().lower() != "codex"
+            str(session.get("provider", "")).strip().lower() != active_provider
             or not isinstance(thread_id, str)
             or not thread_id
         ):
-            reply = "ℹ️ There is no active Codex session to distill."
+            reply = f"ℹ️ There is no active {provider_label} session to distill."
             await message.reply_text(reply)
             log_debug(user_id, "bot", reply)
             return
@@ -635,17 +637,20 @@ async def _cmd_distill(
                 discriminator=self._explicit_distill_discriminator(session),
             )
         except Exception:
-            logger.warning("Explicit Codex distill request could not be recorded")
-            reply = "⚠️ Codex memory distill request could not be recorded."
+            logger.warning(
+                "Explicit %s distill request could not be recorded",
+                provider_label,
+            )
+            reply = f"⚠️ {provider_label} memory distill request could not be recorded."
             await message.reply_text(reply)
             log_debug(user_id, "bot", reply)
             return
 
         if job is None:
-            reply = "⚠️ Codex memory distill is unavailable on this bridge."
+            reply = f"⚠️ {provider_label} memory distill is unavailable on this bridge."
         else:
             reply = (
-                "✅ Codex memory distill request recorded. "
+                f"✅ {provider_label} memory distill request recorded. "
                 "The current session remains active."
             )
         await message.reply_text(reply)