From 77c244a108898c4f6dfe09241315bbf463a195e2 Mon Sep 17 00:00:00 2001 From: jinon86 <247078695+jinon86@users.noreply.github.com> Date: Wed, 5 Aug 2026 12:28:00 +0900 Subject: [PATCH 1/3] feat(memory): add Piri round-trip pipeline --- bridge/__main__.py | 41 ++- bridge/core/bot.py | 23 +- bridge/core/bot_commands.py | 25 +- bridge/core/memory_audience.py | 66 ++++ bridge/core/piri_rpc.py | 6 + bridge/core/piri_runtime.py | 127 +++++++- bridge/core/project_chat_process.py | 4 + bridge/core/provider_capabilities.py | 45 +-- bridge/memory/distill_extraction.py | 7 +- bridge/memory/distill_honcho_worker.py | 12 +- bridge/memory/distill_journal.py | 5 +- bridge/memory/distill_local_sink.py | 2 +- bridge/memory/distill_types.py | 5 +- bridge/memory/distill_worker.py | 6 +- bridge/memory/piri_snapshot.py | 284 ++++++++++++++++++ bridge/tests/test_config_voice_provider.py | 19 +- bridge/tests/test_distill_extraction.py | 9 + bridge/tests/test_distill_honcho_worker.py | 22 ++ bridge/tests/test_distill_journal.py | 17 ++ bridge/tests/test_distill_local_journal.py | 3 +- bridge/tests/test_distill_local_sink.py | 21 +- bridge/tests/test_distill_roundtrip.py | 61 ++++ bridge/tests/test_distill_wiki_worker.py | 21 ++ bridge/tests/test_distill_worker.py | 26 +- bridge/tests/test_piri_runtime.py | 102 +++++++ bridge/tests/test_piri_snapshot.py | 105 +++++++ bridge/tests/test_project_chat_codex.py | 32 ++ bridge/tests/test_provider_capabilities.py | 28 ++ bridge/tests/test_session_composition.py | 40 +++ bridge/tests/test_session_provider.py | 53 +++- bridge/tests/test_shared_group_memory.py | 26 ++ bridge/utils/memory_policy.py | 16 +- claude/hooks/nunchi/piri-feed.sh | 22 +- docs/piri-runtime-contract.md | 25 +- docs/provider-capability-matrix.md | 14 +- .../codex-distill-extraction-v1.schema.json | 5 +- scripts/ccc_codex_memory.py | 32 +- scripts/ccc_codex_memory_test.py | 23 ++ scripts/ccc_doctor.py | 40 ++- scripts/ccc_doctor_bridge_status_test.py | 21 ++ scripts/install-nunchi.sh | 5 +- 41 files changed, 1333 insertions(+), 113 deletions(-) create mode 100644 bridge/memory/piri_snapshot.py create mode 100644 bridge/tests/test_piri_snapshot.py diff --git a/bridge/__main__.py b/bridge/__main__.py index 27132005..7bbd6387 100644 --- a/bridge/__main__.py +++ b/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 @@ -157,13 +157,46 @@ 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.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") agent_runtime = 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, ) telegram_port = telegram_port or Application.builder clock = clock or time @@ -222,7 +255,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 +284,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 +295,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, diff --git a/bridge/core/bot.py b/bridge/core/bot.py index 608de310..afb6c288 100644 --- a/bridge/core/bot.py +++ b/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, ) diff --git a/bridge/core/bot_commands.py b/bridge/core/bot_commands.py index e138cd28..ccc7d78b 100644 --- a/bridge/core/bot_commands.py +++ b/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) diff --git a/bridge/core/memory_audience.py b/bridge/core/memory_audience.py index 172a333a..9e05fd2b 100644 --- a/bridge/core/memory_audience.py +++ b/bridge/core/memory_audience.py @@ -64,6 +64,18 @@ def codex_home(self) -> Path: return self.scope_root / "codex" + @property + def piri_session_dir(self) -> Path: + """Return the Piri transcript directory dedicated to this audience.""" + + return self.scope_root / "piri" / "sessions" + + @property + def piri_bootstrap_home(self) -> Path: + """Return the private materializer home used only for Piri context.""" + + return self.scope_root / "piri" / "bootstrap" + def hook_environment(self, settings: Any) -> dict[str, str]: """Return body-free paths/policy for the existing memory hook stack.""" @@ -148,6 +160,28 @@ def claude_environment(self, settings: Any) -> dict[str, str]: return self.hook_environment(settings) + def piri_environment(self, settings: Any) -> dict[str, str]: + """Return the audience overlay for one Piri RPC process. + + Piri keeps provider credentials and static configuration in the + operator-owned global store, but transcripts and generated memory + context are isolated per opaque audience. The runtime disables Piri's + automatic AGENTS/CLAUDE discovery and appends only the context file + declared here. + """ + + env = self.hook_environment(settings) + env.update( + { + "PIRI_CODING_AGENT_SESSION_DIR": str(self.piri_session_dir), + "CCC_PIRI_BOOTSTRAP_HOME": str(self.piri_bootstrap_home), + "CCC_PIRI_BOOTSTRAP_CONTEXT_FILE": str( + self.piri_bootstrap_home / "AGENTS.md" + ), + } + ) + return env + def _audience_root(settings: Any) -> Path: configured = getattr(settings, "bridge_memory_audience_root", None) @@ -206,6 +240,38 @@ def audience_from_claude_environment( return audience +def audience_from_piri_environment( + settings: Any, environment: Mapping[str, str] | None +) -> MemoryAudience: + """Reconstruct and byte-validate one audience-scoped Piri route.""" + + if environment is None: + raise ValueError("Piri audience-scoped memory requires a route environment") + kind = environment.get("CCC_MEMORY_AUDIENCE") + scope = environment.get("CCC_MEMORY_SCOPE") + if kind == AUDIENCE_SHARED: + if scope != AUDIENCE_SHARED: + raise ValueError("Piri shared memory route is invalid") + elif kind == AUDIENCE_PRIVATE: + suffix = (scope or "").removeprefix("private-") + if ( + not isinstance(scope, str) + or not scope.startswith("private-") + or len(suffix) != 32 + or any(char not in "0123456789abcdef" for char in suffix) + ): + raise ValueError("Piri private memory route is invalid") + else: + raise ValueError("Piri memory audience is invalid") + + assert isinstance(kind, str) and isinstance(scope, str) + audience = MemoryAudience(kind, scope, _audience_root(settings)) + expected = audience.piri_environment(settings) + if dict(environment) != expected: + raise ValueError("Piri audience environment does not match the resolved route") + return audience + + def _read_private_key(path: Path) -> bytes: flags = os.O_RDONLY if hasattr(os, "O_NOFOLLOW"): diff --git a/bridge/core/piri_rpc.py b/bridge/core/piri_rpc.py index e0043b86..757e6cc5 100644 --- a/bridge/core/piri_rpc.py +++ b/bridge/core/piri_rpc.py @@ -84,6 +84,11 @@ async def start(self) -> None: raise self._connection_error if self._process is not None: return + process_options: dict[str, Any] = {} + if os.name == "posix": + # Piri session JSONL and extension state must stay owner-only + # even when the bridge service inherited a permissive umask. + process_options["umask"] = 0o077 process = await asyncio.create_subprocess_exec( *self._command, cwd=self._working_directory, @@ -93,6 +98,7 @@ async def start(self) -> None: stderr=asyncio.subprocess.PIPE, start_new_session=True, limit=STDOUT_BUFFER_LIMIT, + **process_options, ) if process.stdin is None or process.stdout is None or process.stderr is None: process.kill() diff --git a/bridge/core/piri_runtime.py b/bridge/core/piri_runtime.py index fdbef31f..650836b4 100644 --- a/bridge/core/piri_runtime.py +++ b/bridge/core/piri_runtime.py @@ -7,7 +7,9 @@ from contextlib import suppress from dataclasses import dataclass import os +from pathlib import Path import re +import stat from types import MappingProxyType from typing import Any, Protocol, cast @@ -27,6 +29,13 @@ deny_approval, ) from .piri_rpc import PiriRpcProcessClient +from .codex_runtime import _run_codex_memory_bootstrap +from telegram_bot.memory.distill_types import ( + CodexTranscriptSnapshot, + TranscriptBounds, +) +from telegram_bot.memory.piri_snapshot import read_piri_snapshot +from telegram_bot.utils.secure_fs import ensure_private_directory _REASONING_LEVELS = ("minimal", "low", "medium", "high", "xhigh", "max") @@ -68,6 +77,8 @@ def __post_init__(self) -> None: PiriClientFactory = Callable[[PiriLaunchConfig], PiriClient] +PiriMemoryEnvironmentValidator = Callable[[Mapping[str, str]], object] +PiriRouteEnvironmentFactory = Callable[[str, str], Mapping[str, str]] class PiriSession: @@ -218,11 +229,19 @@ def __init__( process_environment: Mapping[str, str] | None = None, model_catalog_directory: str | None = None, auto_confirm_extensions: bool = True, + memory_materializer_path: str | None = None, + memory_bootstrap_timeout_seconds: float = 14.0, + memory_environment_validator: PiriMemoryEnvironmentValidator | None = None, + route_environment_factory: PiriRouteEnvironmentFactory | None = None, ) -> None: if not executable.strip(): raise ValueError("Piri executable must not be empty") if model_catalog_directory is not None and not model_catalog_directory: raise ValueError("Piri model catalog directory must not be empty") + if memory_materializer_path is not None and not memory_materializer_path.strip(): + raise ValueError("Piri memory materializer path must not be empty") + if memory_bootstrap_timeout_seconds <= 0 or memory_bootstrap_timeout_seconds > 30: + raise ValueError("Piri memory bootstrap timeout is invalid") if process_environment is not None: for name, value in process_environment.items(): if not isinstance(name, str) or not name or "\x00" in name: @@ -237,11 +256,32 @@ def __init__( self._auto_confirm_extensions = auto_confirm_extensions self._client_factory = client_factory or self._default_client_factory self._sessions: set[PiriSession] = set() + self._memory_materializer_path = memory_materializer_path + self._memory_bootstrap_timeout_seconds = memory_bootstrap_timeout_seconds + self._memory_environment_validator = memory_environment_validator + self._route_environment_factory = route_environment_factory + self._session_directories: dict[str, Path] = {} async def start_or_resume(self, request: SessionRequest) -> PiriSession: _validate_full_access_request(request) _validate_cli_selection(request) command = [self._executable, "--mode", "rpc", "--approve"] + environment = dict(self._process_environment) + session_directory: Path | None = None + if request.memory_environment is not None: + if ( + self._memory_materializer_path is not None + and self._memory_environment_validator is None + ): + raise ValueError("Piri memory materializer requires a route validator") + if self._memory_environment_validator is not None: + self._memory_environment_validator(request.memory_environment) + environment.update(request.memory_environment) + if self._memory_materializer_path is not None: + session_directory = await self._prepare_memory_bootstrap( + command, + environment, + ) if request.session_id is not None: command.extend(("--session-id", request.session_id)) if request.model is not None: @@ -249,9 +289,6 @@ async def start_or_resume(self, request: SessionRequest) -> PiriSession: if request.effort is not None: command.extend(("--thinking", request.effort)) - environment = dict(self._process_environment) - if request.memory_environment is not None: - environment.update(request.memory_environment) config = PiriLaunchConfig( command=tuple(command), working_directory=request.working_directory, @@ -278,6 +315,8 @@ async def start_or_resume(self, request: SessionRequest) -> PiriSession: if request.session_id is not None and session_id != request.session_id: await client.close() raise RuntimeError("Piri resumed a different session id") + if session_directory is not None: + self._session_directories[session_id] = session_directory session = PiriSession( session_id, @@ -287,6 +326,88 @@ async def start_or_resume(self, request: SessionRequest) -> PiriSession: self._sessions.add(session) return session + async def _prepare_memory_bootstrap( + self, + command: list[str], + environment: dict[str, str], + ) -> Path: + session_value = environment.get("PIRI_CODING_AGENT_SESSION_DIR") + bootstrap_value = environment.get("CCC_PIRI_BOOTSTRAP_HOME") + context_value = environment.get("CCC_PIRI_BOOTSTRAP_CONTEXT_FILE") + if not session_value or not bootstrap_value or not context_value: + raise ValueError("Piri memory route is incomplete") + session_directory = Path(session_value) + bootstrap_home = Path(bootstrap_value) + context_file = Path(context_value) + if context_file != bootstrap_home / "AGENTS.md": + raise ValueError("Piri memory context path is invalid") + ensure_private_directory(session_directory) + ensure_private_directory(bootstrap_home) + bootstrap_environment = dict(environment) + bootstrap_environment["CODEX_HOME"] = str(bootstrap_home) + bootstrap_environment["CODEX_SQLITE_HOME"] = str(bootstrap_home) + bootstrap_environment["CCC_MEMORY_MATERIALIZER_PROVIDER"] = "piri" + await _run_codex_memory_bootstrap( + self._memory_materializer_path or "", + timeout_seconds=self._memory_bootstrap_timeout_seconds, + environment=bootstrap_environment, + ) + self._validate_memory_context(context_file) + command.extend(("--no-context-files", "--append-system-prompt", str(context_file))) + return session_directory + + @staticmethod + def _validate_memory_context(path: Path) -> None: + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError: + raise RuntimeError("Piri memory bootstrap unavailable") from None + try: + metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink != 1 + or (hasattr(os, "geteuid") and metadata.st_uid != os.geteuid()) + or stat.S_IMODE(metadata.st_mode) != 0o600 + or metadata.st_size > 1024 * 1024 + ): + raise RuntimeError("Piri memory bootstrap unavailable") + finally: + os.close(descriptor) + + async def read_session_snapshot( + self, + session_id: str, + *, + bounds: TranscriptBounds, + memory_audience: str | None = None, + memory_scope: str | None = None, + ) -> CodexTranscriptSnapshot: + session_directory: Path | None = None + if memory_audience is not None or memory_scope is not None: + if ( + not isinstance(memory_audience, str) + or not isinstance(memory_scope, str) + or self._route_environment_factory is None + ): + raise ValueError("Piri snapshot memory route is invalid") + route = self._route_environment_factory(memory_audience, memory_scope) + value = route.get("PIRI_CODING_AGENT_SESSION_DIR") + if not isinstance(value, str) or not value: + raise ValueError("Piri snapshot memory route is incomplete") + session_directory = Path(value) + else: + session_directory = self._session_directories.get(session_id) + if session_directory is None: + raise ValueError("Piri snapshot session route is unavailable") + return await asyncio.to_thread( + read_piri_snapshot, + session_directory, + session_id, + bounds=bounds, + ) + async def list_models(self) -> Sequence[ModelInfo]: config = PiriLaunchConfig( command=(self._executable, "--mode", "rpc", "--approve", "--no-session"), diff --git a/bridge/core/project_chat_process.py b/bridge/core/project_chat_process.py index b96152d3..3470b58e 100644 --- a/bridge/core/project_chat_process.py +++ b/bridge/core/project_chat_process.py @@ -696,6 +696,10 @@ async def _process_agent_message( # noqa: C901 -- #348 baseline hotspot memory_environment = audience.claude_environment( self._config ) + elif provider == "piri": + memory_environment = audience.piri_environment( + self._config + ) session = await runtime.start_or_resume( SessionRequest( working_directory=str(self.project_root), diff --git a/bridge/core/provider_capabilities.py b/bridge/core/provider_capabilities.py index a514591b..50cad076 100644 --- a/bridge/core/provider_capabilities.py +++ b/bridge/core/provider_capabilities.py @@ -562,9 +562,10 @@ def _unknown(reason: str, *dependencies: str) -> CapabilityStatus: "The AGENTS.override.md materializer runs before thread start/resume; " "promoted after the 2026-07-15 live gate (#419)." ), - piri=_degraded( - "Piri starts in the project directory and can consume project context, " - "but the ccc audience-scoped memory materializer is intentionally disabled." + piri=_supported( + "Audience-scoped Piri sessions use isolated transcript directories, disable " + "automatic AGENTS/CLAUDE discovery, run the bounded ccc materializer, and " + "append only the generated scope-local AGENTS.md context." ), ), _axis( @@ -582,9 +583,10 @@ def _unknown(reason: str, *dependencies: str) -> CapabilityStatus: "not treated as compaction. Provider compaction checkpoint/reinjection " "therefore remains unverified." ), - piri=_unsupported( - "ccc-node has no Piri compaction checkpoint or post-compaction memory " - "reinjection hook." + piri=_degraded( + "Cold start/resume refreshes and re-appends the scoped snapshot, but Piri " + "exposes no ccc compaction lifecycle hook for an explicit mid-session " + "checkpoint or reinjection proof." ), ), _axis( @@ -603,12 +605,10 @@ def _unknown(reason: str, *dependencies: str) -> CapabilityStatus: "cost gates are body-free; Wiki candidates enter a local human-review " "queue and Honcho facts use an owner-only retrying outbox.", ), - piri=_unsupported( - "Piri sessions are not read by the Codex/Claude distill journal, " - "so the local/Honcho/Wiki distill sinks stay unwired. (A " - "provider-neutral nunchi peer-facts extractor, " - "`hooks/nunchi/piri-feed.sh` via `install-nunchi.sh --piri`, does " - "feed the nunchi DB from Piri sessions.)" + piri=_supported( + "Session-reset, explicit, checkpoint, and shutdown triggers enter the " + "provider-neutral journal. A secure bounded Piri JSONL snapshot preserves " + "source provider provenance through the isolated extractor and all sinks." ), ), _axis( @@ -624,8 +624,9 @@ def _unknown(reason: str, *dependencies: str) -> CapabilityStatus: "Supported session-reset triggers bind an opaque audience route, and " "an independently leased worker writes replay-safe local facts/resume." ), - piri=_unsupported( - "No Piri write-back extractor feeds the replay-safe local memory sink." + piri=_supported( + "Audience-routed Piri jobs use the same independently leased replay-safe " + "local facts/resume sink while retaining provider=piri provenance." ), ), _axis( @@ -644,8 +645,9 @@ def _unknown(reason: str, *dependencies: str) -> CapabilityStatus: "physically distinct Honcho workspaces; unscoped jobs fail closed in " "that mode." ), - piri=_unsupported( - "No Piri write-back extractor feeds the Honcho outbox." + piri=_supported( + "Validated Piri facts use the same owner-only scope-partitioned Honcho " + "outbox, idempotency keys, retry leases, and distinct workspaces." ), ), _axis( @@ -661,8 +663,9 @@ def _unknown(reason: str, *dependencies: str) -> CapabilityStatus: "Validated candidates are atomically queued in owner-only per-job records; " "the sink performs no Wiki write, branch, PR, or merge." ), - piri=_unsupported( - "No Piri write-back extractor feeds the human-gated Wiki candidate queue." + piri=_supported( + "Validated Piri candidates enter the same owner-only human-review queue; " + "the sink still performs no Wiki write, branch, PR, or merge." ), ), _axis( @@ -681,9 +684,9 @@ def _unknown(reason: str, *dependencies: str) -> CapabilityStatus: "one durable fact exactly once; local, Wiki-candidate, and Honcho " "sink states remain independently replayable (#465)." ), - piri=_unsupported( - "A Piri session A to durable write-back to isolated session B round-trip " - "does not exist because Piri write-back is not implemented." + piri=_supported( + "The hermetic audience-scoped Piri A→snapshot→distill→local index→B " + "bootstrap test recalls one durable fact with provider=piri provenance." ), ), _axis( diff --git a/bridge/memory/distill_extraction.py b/bridge/memory/distill_extraction.py index aaa4b1bf..c2b8ad0a 100644 --- a/bridge/memory/distill_extraction.py +++ b/bridge/memory/distill_extraction.py @@ -124,7 +124,7 @@ def validate_text(cls, value: str) -> str: class DistillExtractionInput(_StrictModel): schema_version: Literal[1] - provider: Literal["codex"] + provider: Literal["codex", "piri"] content_trust: Literal["untrusted"] source_thread_hash: str = Field(pattern=_SHA256_RE.pattern) trigger: DistillTrigger @@ -163,7 +163,7 @@ def model_post_init(self, __context: Any) -> None: class DistillProvenance(_StrictModel): - provider: Literal["codex"] + provider: Literal["codex", "piri"] source_thread_hash: str = Field(pattern=_SHA256_RE.pattern) trigger: DistillTrigger distilled_at: str = Field(json_schema_extra={"format": "date-time"}) @@ -340,6 +340,7 @@ def build_extraction_input( snapshot: CodexTranscriptSnapshot, *, trigger: DistillTrigger, + provider: Literal["codex", "piri"] = "codex", ) -> DistillExtractionInput: """Normalize a bounded snapshot and redact credentials before provider use.""" if not isinstance(snapshot, CodexTranscriptSnapshot): @@ -352,7 +353,7 @@ def build_extraction_input( ) return DistillExtractionInput( schema_version=DISTILL_EXTRACTION_SCHEMA_VERSION, - provider="codex", + provider=provider, content_trust="untrusted", source_thread_hash=snapshot.thread_hash, trigger=trigger, diff --git a/bridge/memory/distill_honcho_worker.py b/bridge/memory/distill_honcho_worker.py index a39ad3eb..70c5ddce 100644 --- a/bridge/memory/distill_honcho_worker.py +++ b/bridge/memory/distill_honcho_worker.py @@ -145,6 +145,10 @@ def send(self, record: dict[str, object]) -> None: or not isinstance(provenance, dict) ): raise HonchoDeliveryError("honcho_record_invalid", terminal=True) + source_provider = provenance.get("provider") + if source_provider not in {"codex", "piri"}: + raise HonchoDeliveryError("honcho_record_invalid", terminal=True) + source = f"{source_provider}-distill" try: validate_memory_route(memory_audience, memory_scope) except ValueError: @@ -164,7 +168,7 @@ def send(self, record: dict[str, object]) -> None: payload={ "id": session_id, "metadata": { - "source": "codex-distill", + "source": source, "node": self._node_label, **route_metadata, }, @@ -182,9 +186,9 @@ def send(self, record: dict[str, object]) -> None: payload={ "messages": [{ "peer_id": peer, - "content": "[codex distill]\n" + content, + "content": f"[{source.replace('-', ' ')}]\n" + content, "metadata": { - "source": "codex-distill", "node": self._node_label, + "source": source, "node": self._node_label, "idempotency_key": key, "provenance": provenance, "facts": facts, **route_metadata, }, @@ -256,7 +260,7 @@ def record( record: dict[str, object] = { "schema_version": output.schema_version, "idempotency_key": f"ccc-distill-{job_id}", - "session_id": f"codex-distill-{job_id[:24]}", + "session_id": f"{provenance.provider}-distill-{job_id[:24]}", "provenance": { "provider": provenance.provider, "source_thread_hash": provenance.source_thread_hash, diff --git a/bridge/memory/distill_journal.py b/bridge/memory/distill_journal.py index 0f0b2b23..acb15b81 100644 --- a/bridge/memory/distill_journal.py +++ b/bridge/memory/distill_journal.py @@ -23,6 +23,7 @@ DistillLocalSinkStatus, DistillTrigger, DistillWikiSinkStatus, + DISTILL_PROVIDERS, validate_memory_route, ) from .distill_extraction import ( @@ -146,8 +147,8 @@ def enqueue_once( memory_scope: str | None = None, now: datetime | None = None, ) -> DistillJob: - if provider != "codex": - raise ValueError("distill journal accepts Codex jobs only") + if provider not in DISTILL_PROVIDERS: + raise ValueError("distill journal provider is unsupported") if not isinstance(thread_id, str) or not thread_id: raise ValueError("thread_id must not be empty") if not discriminator: diff --git a/bridge/memory/distill_local_sink.py b/bridge/memory/distill_local_sink.py index 28dcd47e..f3358f47 100644 --- a/bridge/memory/distill_local_sink.py +++ b/bridge/memory/distill_local_sink.py @@ -263,7 +263,7 @@ def transform( LocalMemoryTransaction(self.state_dir).commit( transform, - provider="codex", + provider=output.provenance.provider, actor="distill", tool="local-memory-sink", session=job_id, diff --git a/bridge/memory/distill_types.py b/bridge/memory/distill_types.py index a32329eb..4406befb 100644 --- a/bridge/memory/distill_types.py +++ b/bridge/memory/distill_types.py @@ -13,6 +13,7 @@ _SAFE_ERROR_CODE_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") _PRIVATE_MEMORY_SCOPE_RE = re.compile(r"^private-[0-9a-f]{32}$") _DISTILL_MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +DISTILL_PROVIDERS = frozenset({"codex", "piri"}) class DistillTrigger(str, Enum): @@ -390,8 +391,8 @@ class DistillJob: def __post_init__(self) -> None: if not _SHA256_RE.fullmatch(self.job_id): raise ValueError("job_id must be a SHA-256 hex digest") - if self.provider != "codex": - raise ValueError("distill jobs support the Codex provider only") + if self.provider not in DISTILL_PROVIDERS: + raise ValueError("distill job provider is unsupported") if not isinstance(self.thread_id, str) or not self.thread_id: raise ValueError("invalid distill job thread identity") expected_thread_hash = hashlib.sha256(self.thread_id.encode("utf-8")).hexdigest() diff --git a/bridge/memory/distill_worker.py b/bridge/memory/distill_worker.py index 7656ba88..1e5ab915 100644 --- a/bridge/memory/distill_worker.py +++ b/bridge/memory/distill_worker.py @@ -257,7 +257,11 @@ async def extract_once(self, *, job_id: str) -> DistillJob: terminal=True, ) try: - extraction_input = build_extraction_input(snapshot, trigger=claimed.trigger) + extraction_input = build_extraction_input( + snapshot, + trigger=claimed.trigger, + provider=claimed.provider, + ) except (TypeError, ValueError): self._refund_unused_reservation(reservation) return await self._fail( diff --git a/bridge/memory/piri_snapshot.py b/bridge/memory/piri_snapshot.py new file mode 100644 index 00000000..8efc23be --- /dev/null +++ b/bridge/memory/piri_snapshot.py @@ -0,0 +1,284 @@ +"""Secure, bounded snapshots for audience-scoped Piri JSONL sessions.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from dataclasses import dataclass +import hashlib +import json +import os +from pathlib import Path +import stat +from typing import Any, Literal, Mapping + +from .distill_types import ( + CodexTranscriptSnapshot, + TranscriptBounds, + TranscriptMessage, +) + +_MAX_DIRECTORY_ENTRIES = 4096 +_MAX_HEADER_BYTES = 64 * 1024 +_MAX_SCAN_BYTES = 8 * 1024 * 1024 + + +@dataclass(frozen=True, slots=True) +class _Candidate: + role: Literal["user", "assistant"] + text: str + timestamp: str + entry_id: str | None + + +def _timestamp(value: object, *, fallback: datetime) -> tuple[str, datetime]: + parsed: datetime | None = None + if isinstance(value, str) and value: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + parsed = None + elif isinstance(value, (int, float)) and not isinstance(value, bool): + seconds = float(value) / 1000.0 if abs(float(value)) >= 10**11 else float(value) + try: + parsed = datetime.fromtimestamp(seconds, tz=timezone.utc) + except (OverflowError, OSError, ValueError): + parsed = None + if parsed is None: + parsed = fallback + elif parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + parsed = parsed.astimezone(timezone.utc) + return parsed.isoformat().replace("+00:00", "Z"), parsed + + +def _text_content(content: object) -> str: + if isinstance(content, str): + return content.strip() + if not isinstance(content, list): + return "" + parts: list[str] = [] + for block in content: + if not isinstance(block, Mapping) or block.get("type") != "text": + continue + text = block.get("text") + if isinstance(text, str) and text: + parts.append(text) + return " ".join(parts).strip() + + +def _bounded_text(value: str, limit: int) -> tuple[str, bool]: + encoded = value.encode("utf-8") + if len(encoded) <= limit: + return value, False + return encoded[:limit].decode("utf-8", errors="ignore").rstrip(), True + + +def _validate_directory(path: Path) -> None: + metadata = path.lstat() + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): + raise ValueError("Piri session directory is unsafe") + if hasattr(os, "geteuid") and metadata.st_uid != os.geteuid(): + raise ValueError("Piri session directory owner is unsafe") + if stat.S_IMODE(metadata.st_mode) & 0o077: + raise ValueError("Piri session directory permissions are unsafe") + + +def _open_session(path: Path) -> tuple[int, os.stat_result]: + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + metadata = os.fstat(descriptor) + unsafe = ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink != 1 + or (hasattr(os, "geteuid") and metadata.st_uid != os.geteuid()) + or stat.S_IMODE(metadata.st_mode) & 0o077 + ) + if unsafe: + os.close(descriptor) + raise ValueError("Piri session file is unsafe") + return descriptor, metadata + + +def _session_path(session_dir: Path, session_id: str) -> Path | None: + if not session_dir.exists(): + return None + _validate_directory(session_dir) + suffix = f"_{session_id}.jsonl" + exact = f"{session_id}.jsonl" + candidates: list[Path] = [] + with os.scandir(session_dir) as entries: + for index, entry in enumerate(entries, start=1): + if index > _MAX_DIRECTORY_ENTRIES: + raise ValueError("Piri session directory exceeds its safe bound") + if entry.name != exact and not entry.name.endswith(suffix): + continue + if not entry.is_file(follow_symlinks=False): + raise ValueError("Piri session entry is unsafe") + candidates.append(session_dir / entry.name) + if not candidates: + return None + if len(candidates) != 1: + raise ValueError("Piri session id is ambiguous") + return candidates[0] + + +def _read_payload(path: Path, session_id: str, limits: TranscriptBounds) -> tuple[bytes, os.stat_result]: + descriptor, metadata = _open_session(path) + try: + header = os.pread(descriptor, _MAX_HEADER_BYTES + 1, 0) + first_line = header.splitlines()[0] if header else b"" + if len(first_line) > _MAX_HEADER_BYTES: + raise ValueError("Piri session header exceeds its safe bound") + try: + value = json.loads(first_line) + except (UnicodeError, ValueError): + raise ValueError("Piri session header is invalid") from None + if ( + not isinstance(value, Mapping) + or value.get("type") != "session" + or value.get("id") != session_id + ): + raise ValueError("Piri session identity does not match") + scan_bytes = min( + _MAX_SCAN_BYTES, + max(1024 * 1024, limits.max_bytes * 16, limits.max_message_bytes * limits.max_items), + ) + offset = max(0, metadata.st_size - scan_bytes) + payload = os.pread(descriptor, metadata.st_size - offset, offset) + if offset: + newline = payload.find(b"\n") + payload = b"" if newline < 0 else payload[newline + 1 :] + return payload, metadata + finally: + os.close(descriptor) + + +def _parse_candidate( + raw_line: bytes, + *, + fallback: datetime, + captured: datetime, + limits: TranscriptBounds, +) -> tuple[_Candidate | None, bool]: + try: + entry: Any = json.loads(raw_line) + except (UnicodeError, ValueError): + return None, True + if not isinstance(entry, Mapping) or entry.get("type") != "message": + return None, False + message = entry.get("message") + if not isinstance(message, Mapping): + return None, False + raw_role = message.get("role") + if raw_role not in {"user", "assistant"}: + return None, False + role: Literal["user", "assistant"] = raw_role + text = _text_content(message.get("content")) + if not text: + return None, False + timestamp, parsed_time = _timestamp(entry.get("timestamp"), fallback=fallback) + if (captured - parsed_time).total_seconds() > limits.max_age_seconds: + return None, True + text, was_truncated = _bounded_text(text, limits.max_message_bytes) + raw_id = entry.get("id") + entry_id = raw_id if isinstance(raw_id, str) and raw_id else None + return _Candidate(role, text, timestamp, entry_id), was_truncated + + +def _collect_messages( + payload: bytes, + *, + metadata: os.stat_result, + limits: TranscriptBounds, + captured: datetime, +) -> tuple[tuple[TranscriptMessage, ...], int, str | None, bool]: + fallback = datetime.fromtimestamp(metadata.st_mtime, tz=timezone.utc) + newest: list[TranscriptMessage] = [] + byte_count = 0 + user_turns = 0 + items_seen = 0 + last_turn_id: str | None = None + truncated = metadata.st_size > len(payload) + for raw_line in reversed(payload.splitlines()): + if items_seen >= limits.max_items: + truncated = True + break + items_seen += 1 + candidate, candidate_truncated = _parse_candidate( + raw_line, + fallback=fallback, + captured=captured, + limits=limits, + ) + truncated = truncated or candidate_truncated + if candidate is None: + continue + if candidate.role == "user": + user_turns += 1 + if user_turns > limits.max_turns: + truncated = True + break + remaining = limits.max_bytes - byte_count + if remaining <= 0: + truncated = True + break + text, was_truncated = _bounded_text(candidate.text, remaining) + truncated = truncated or was_truncated + if not text: + break + newest.append(TranscriptMessage(candidate.role, text, candidate.timestamp)) + byte_count += len(text.encode("utf-8")) + if last_turn_id is None: + last_turn_id = candidate.entry_id + if len(newest) >= limits.max_messages: + truncated = True + break + return tuple(reversed(newest)), byte_count, last_turn_id, truncated + + +def read_piri_snapshot( + session_dir: Path, + session_id: str, + *, + bounds: TranscriptBounds, + now: datetime | None = None, +) -> CodexTranscriptSnapshot: + """Return newest user/assistant messages without following provider paths.""" + + if not session_id: + raise ValueError("Piri session id must not be empty") + captured = now or datetime.now(timezone.utc) + if captured.tzinfo is None: + captured = captured.replace(tzinfo=timezone.utc) + captured = captured.astimezone(timezone.utc) + thread_hash = hashlib.sha256(session_id.encode("utf-8")).hexdigest() + path = _session_path(Path(session_dir), session_id) + if path is None: + return CodexTranscriptSnapshot( + thread_hash=thread_hash, + last_turn_id=None, + messages=(), + byte_count=0, + truncated=False, + captured_at=captured.isoformat().replace("+00:00", "Z"), + ) + + payload, metadata = _read_payload(path, session_id, bounds) + messages, byte_count, last_turn_id, truncated = _collect_messages( + payload, + metadata=metadata, + limits=bounds, + captured=captured, + ) + + return CodexTranscriptSnapshot( + thread_hash=thread_hash, + last_turn_id=last_turn_id, + messages=messages, + byte_count=byte_count, + truncated=truncated, + captured_at=captured.isoformat().replace("+00:00", "Z"), + ) + + +__all__ = ["read_piri_snapshot"] diff --git a/bridge/tests/test_config_voice_provider.py b/bridge/tests/test_config_voice_provider.py index 75a6ba96..523297ae 100644 --- a/bridge/tests/test_config_voice_provider.py +++ b/bridge/tests/test_config_voice_provider.py @@ -138,17 +138,14 @@ def test_shared_group_memory_and_image_guards_are_explicit_opt_ins(self): _env_file=None, ) self.assertEqual(codex_curated.agent_provider, "codex") - with self.assertRaisesRegex( - ValidationError, - "audience-scoped memory cannot run with the Piri provider", - ): - module.Config( - telegram_bot_token="123456:abc", - CCC_TELEGRAM_SESSION_SCOPE="shared-groups", - CCC_BRIDGE_MEMORY_MODE="audience-scoped", - CCC_AGENT_PROVIDER="piri", - _env_file=None, - ) + piri_audience = module.Config( + telegram_bot_token="123456:abc", + CCC_TELEGRAM_SESSION_SCOPE="shared-groups", + CCC_BRIDGE_MEMORY_MODE="audience-scoped", + CCC_AGENT_PROVIDER="piri", + _env_file=None, + ) + self.assertEqual(piri_audience.agent_provider, "piri") piri_curated = module.Config( telegram_bot_token="123456:abc", CCC_BRIDGE_MEMORY_MODE="curated", diff --git a/bridge/tests/test_distill_extraction.py b/bridge/tests/test_distill_extraction.py index 61a60eb5..8a2123d9 100644 --- a/bridge/tests/test_distill_extraction.py +++ b/bridge/tests/test_distill_extraction.py @@ -136,6 +136,15 @@ def test_same_snapshot_and_trigger_produce_byte_identical_redacted_input() -> No assert first.byte_count == len(first.messages[0].text.encode("utf-8")) +def test_piri_snapshot_keeps_source_provider_in_extraction_contract() -> None: + extraction = build_extraction_input( + snapshot(), + trigger=DistillTrigger.EXPLICIT, + provider="piri", + ) + assert extraction.provider == "piri" + + @pytest.mark.parametrize( ("mutation", "match"), [ diff --git a/bridge/tests/test_distill_honcho_worker.py b/bridge/tests/test_distill_honcho_worker.py index 749d5a76..5a3d1fd9 100644 --- a/bridge/tests/test_distill_honcho_worker.py +++ b/bridge/tests/test_distill_honcho_worker.py @@ -69,6 +69,28 @@ async def test_worker_delivers_stable_body_safe_record_and_acks_outbox( assert list((outbox / str(job.memory_scope)).glob("*.json")) == [] +@pytest.mark.anyio +async def test_worker_preserves_piri_provenance_in_honcho_record( + tmp_path: Path, +) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = await extracted_job(journal, provider="piri") + sender = RecordingSender() + worker = CodexDistillHonchoSinkWorker( + journal, + outbox_dir=tmp_path / "honcho-outbox", + sender=sender, + owner_token="piri-honcho-worker", + ) + + result = await worker.write_once(job_id=job.job_id) + + assert result.honcho_sink_status is DistillHonchoSinkStatus.DONE + assert sender.records[0]["session_id"] == f"piri-distill-{job.job_id[:24]}" + assert sender.records[0]["provenance"]["provider"] == "piri" + + @pytest.mark.anyio async def test_delivery_failure_preserves_one_outbox_and_retries_without_extraction( tmp_path: Path, diff --git a/bridge/tests/test_distill_journal.py b/bridge/tests/test_distill_journal.py index c7fc9b07..0c1231d9 100644 --- a/bridge/tests/test_distill_journal.py +++ b/bridge/tests/test_distill_journal.py @@ -58,6 +58,23 @@ def enqueue(): assert stat.S_IMODE(journal.job_path(jobs[0].job_id).stat().st_mode) == 0o600 +def test_piri_provider_jobs_are_first_class_and_separate_from_codex(tmp_path: Path) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + piri = journal.enqueue_once( + provider="piri", + thread_id="same-session", + trigger=DistillTrigger.EXPLICIT, + ) + codex = journal.enqueue_once( + provider="codex", + thread_id="same-session", + trigger=DistillTrigger.EXPLICIT, + ) + assert piri.provider == "piri" + assert piri.job_id != codex.job_id + + def test_job_key_is_cross_trigger_idempotent_and_binds_discriminator_and_schema( tmp_path: Path, ) -> None: diff --git a/bridge/tests/test_distill_local_journal.py b/bridge/tests/test_distill_local_journal.py index f0045ef1..7e5d41f9 100644 --- a/bridge/tests/test_distill_local_journal.py +++ b/bridge/tests/test_distill_local_journal.py @@ -29,9 +29,10 @@ async def extracted_job( memory_scope: str | None = PRIVATE_SCOPE, wiki_enabled: bool = True, honcho_enabled: bool = True, + provider: str = "codex", ) -> DistillJob: queued = journal.enqueue_once( - provider="codex", + provider=provider, thread_id="thread-local-journal", trigger=DistillTrigger.NEW_COMMAND, memory_audience=memory_audience, diff --git a/bridge/tests/test_distill_local_sink.py b/bridge/tests/test_distill_local_sink.py index cdbcb068..95056e7b 100644 --- a/bridge/tests/test_distill_local_sink.py +++ b/bridge/tests/test_distill_local_sink.py @@ -23,12 +23,13 @@ def extraction_output( *, fact_text: str = "The user prefers focused pull requests.", last_activity: str = "Implemented a bounded local sink.", + provider: str = "codex", ) -> DistillExtractionOutput: return DistillExtractionOutput.model_validate( { "schema_version": 1, "provenance": { - "provider": "codex", + "provider": provider, "source_thread_hash": THREAD_HASH, "trigger": "new_command", "distilled_at": "2026-07-22T08:00:00Z", @@ -97,6 +98,24 @@ def test_writes_bounded_private_facts_and_resume_with_hashed_provenance( assert stat.S_IMODE(resume_path.stat().st_mode) == 0o600 +def test_piri_provenance_reaches_facts_resume_and_rollback_ledger( + tmp_path: Path, +) -> None: + state_dir = tmp_path / "state" + sink = CodexLocalMemorySink(state_dir, audience="private") + + sink.write(extraction_output(provider="piri"), job_id=JOB_ID) + + facts = read_facts(state_dir / "memory-facts.jsonl") + assert facts[0]["source"]["provider"] == "piri" + assert "provider=piri" in (state_dir / "resume.md").read_text() + action_id = (state_dir / "memory-rollback" / "HEAD").read_text().strip() + manifest = json.loads( + (state_dir / "memory-rollback" / "actions" / action_id / "manifest.json").read_text() + ) + assert manifest["provider"] == "piri" + + @pytest.mark.anyio async def test_ten_concurrent_replays_append_each_fact_once(tmp_path: Path) -> None: state_dir = tmp_path / "state" diff --git a/bridge/tests/test_distill_roundtrip.py b/bridge/tests/test_distill_roundtrip.py index 7aa60f3e..87b7c67b 100644 --- a/bridge/tests/test_distill_roundtrip.py +++ b/bridge/tests/test_distill_roundtrip.py @@ -83,3 +83,64 @@ async def test_thread_a_fact_appears_once_in_isolated_thread_b_snapshot( assert snapshot.count(FACT) == 1 assert "Honcho disabled" in snapshot assert "Family Wiki disabled" in snapshot + + +@pytest.mark.anyio +async def test_piri_thread_a_fact_appears_in_next_audience_bootstrap( + tmp_path: Path, +) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = await extracted_job(journal, provider="piri") + audience_root = tmp_path / "audiences" + worker = CodexDistillLocalSinkWorker( + journal, + audience_root=audience_root, + owner_token="piri-roundtrip-local-worker", + indexer_path=ROOT / "scripts" / "ccc-memory-index.sh", + ) + + written = await worker.write_once(job_id=job.job_id) + + assert written.local_sink_status is DistillLocalSinkStatus.DONE + scope = str(job.memory_scope) + audience = MemoryAudience("private", scope, audience_root) + local_facts = (audience.state_dir / "memory-facts.jsonl").read_text() + assert '"provider":"piri"' in local_facts + settings = SimpleNamespace( + claude_settings_path=tmp_path / "legacy" / ".claude" / "settings.json", + honcho_memory_enabled=False, + honcho_config_path=tmp_path / ".hermes" / "honcho.json", + ) + environment = os.environ.copy() + environment.update(audience.piri_environment(settings)) + environment.update( + { + "HOME": str(tmp_path / "home"), + "PROJECT_ROOT": str(ROOT), + "CODEX_HOME": str(audience.piri_bootstrap_home), + "CODEX_SQLITE_HOME": str(audience.piri_bootstrap_home), + "CCC_MEMORY_MATERIALIZER_PROVIDER": "piri", + "CCC_CODEX_MEMORY_LOADER": str(ROOT / "claude" / "hooks" / "load-memory.sh"), + "CCC_HOOK_DIR": str(ROOT / "claude" / "hooks"), + "CCC_MEMORY_TOOLS_DIR": str(ROOT / "scripts"), + "CCC_MEMORY_NO_REFRESH": "1", + "CCC_LOCAL_MEMORY_ENABLED": "1", + "CCC_CODEX_MEMORY_MAX_BYTES": "8192", + "CCC_CODEX_AGENTS_BUDGET_BYTES": "16384", + } + ) + + completed = subprocess.run( + [sys.executable, str(ROOT / "scripts" / "ccc_codex_memory.py"), "materialize", "--json"], + env=environment, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=20, + check=False, + ) + + assert completed.returncode == 0, completed.stderr + context = (audience.piri_bootstrap_home / "AGENTS.md").read_text() + assert context.count(FACT) == 1 diff --git a/bridge/tests/test_distill_wiki_worker.py b/bridge/tests/test_distill_wiki_worker.py index 35a79e39..0a9ac396 100644 --- a/bridge/tests/test_distill_wiki_worker.py +++ b/bridge/tests/test_distill_wiki_worker.py @@ -29,12 +29,14 @@ async def wiki_job( *, memory_audience: str | None = None, memory_scope: str | None = None, + provider: str = "codex", ): # type: ignore[no-untyped-def] snapshot_done = fixtures.snapshot_done_job( journal, thread_id="thread-wiki-worker", memory_audience=memory_audience, memory_scope=memory_scope, + provider=provider, ) return await CodexDistillExtractionWorker( journal, @@ -63,6 +65,25 @@ async def test_worker_queues_validated_candidate_for_human_review(tmp_path: Path assert record["candidates"][0]["suggested_path"].startswith("pages/nodes/") +@pytest.mark.anyio +async def test_worker_preserves_piri_provenance_in_wiki_candidate( + tmp_path: Path, +) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = await wiki_job(journal, provider="piri") + queue = tmp_path / "wiki-candidates" + worker = CodexDistillWikiSinkWorker( + journal, queue_dir=queue, owner_token="piri-wiki-worker" + ) + + result = await worker.write_once(job_id=job.job_id) + + assert result.wiki_sink_status is DistillWikiSinkStatus.DONE + record = json.loads((queue / f"{job.job_id}.json").read_text()) + assert record["provenance"]["provider"] == "piri" + + @pytest.mark.anyio @pytest.mark.parametrize( ("memory_audience", "scope"), diff --git a/bridge/tests/test_distill_worker.py b/bridge/tests/test_distill_worker.py index 5c7245df..f133df11 100644 --- a/bridge/tests/test_distill_worker.py +++ b/bridge/tests/test_distill_worker.py @@ -88,9 +88,10 @@ def snapshot_done_job( text: str = "harmless durable fact", memory_audience: str | None = None, memory_scope: str | None = None, + provider: str = "codex", ) -> DistillJob: queued = journal.enqueue_once( - provider="codex", + provider=provider, thread_id=thread_id, trigger=DistillTrigger.NEW_COMMAND, memory_audience=memory_audience, @@ -123,6 +124,29 @@ async def extract( return output_for(extraction_input) +@pytest.mark.anyio +async def test_piri_job_preserves_source_provider_through_extraction( + tmp_path: Path, +) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = snapshot_done_job(journal, provider="piri") + backend = SuccessfulBackend() + + result = await CodexDistillExtractionWorker( + journal, + backend, + owner_token="piri-extract-worker", + usage_meter=None, + ).extract_once(job_id=job.job_id) + + assert result.status is DistillJobStatus.EXTRACTION_DONE + assert backend.calls[0].provider == "piri" + output = journal.get_extraction_output(job.job_id) + assert output is not None + assert output.provenance.provider == "piri" + + def wiki_output_for( extraction_input: DistillExtractionInput, ) -> DistillExtractionOutput: diff --git a/bridge/tests/test_piri_runtime.py b/bridge/tests/test_piri_runtime.py index b1a991e3..a2d90b33 100644 --- a/bridge/tests/test_piri_runtime.py +++ b/bridge/tests/test_piri_runtime.py @@ -4,11 +4,14 @@ import asyncio from collections.abc import Mapping, Sequence +import os from pathlib import Path +import stat import sys import tempfile from typing import TYPE_CHECKING, Any import unittest +from unittest.mock import patch if TYPE_CHECKING: from core.agent_runtime import AgentEvent @@ -140,6 +143,96 @@ async def test_new_session_uses_unrestricted_process_contract(self) -> None: self.assertEqual(config.environment["MEMORY"], "two") self.assertTrue(config.auto_confirm_extensions) + async def test_audience_memory_is_materialized_and_isolated_from_context_discovery( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + session_dir = root / "sessions" + bootstrap_home = root / "bootstrap" + context_file = bootstrap_home / "AGENTS.md" + memory_environment = { + "PIRI_CODING_AGENT_SESSION_DIR": str(session_dir), + "CCC_PIRI_BOOTSTRAP_HOME": str(bootstrap_home), + "CCC_PIRI_BOOTSTRAP_CONTEXT_FILE": str(context_file), + } + + async def materialize( + _path: str, *, timeout_seconds: float, environment: Mapping[str, str] + ) -> None: + self.assertEqual(timeout_seconds, 3.0) + self.assertEqual(environment["CODEX_HOME"], str(bootstrap_home)) + self.assertEqual(environment["CCC_MEMORY_MATERIALIZER_PROVIDER"], "piri") + context_file.write_text("scoped memory", encoding="utf-8") + context_file.chmod(0o600) + + runtime = PiriRuntime( + executable="/opt/piri/bin/piri", + client_factory=self.factory, + process_environment={"BASE": "one"}, + memory_materializer_path="/materializer", + memory_bootstrap_timeout_seconds=3.0, + memory_environment_validator=lambda value: self.assertEqual( + dict(value), memory_environment + ), + ) + with patch( + "telegram_bot.core.piri_runtime._run_codex_memory_bootstrap", + side_effect=materialize, + ): + session = await runtime.start_or_resume( + SessionRequest( + working_directory="/workspace/project", + memory_environment=memory_environment, + ) + ) + + config = self.factory.clients[-1].config + self.assertIn("--no-context-files", config.command) + self.assertEqual( + config.command[-2:], + ("--append-system-prompt", str(context_file)), + ) + self.assertEqual( + config.environment["PIRI_CODING_AGENT_SESSION_DIR"], + str(session_dir), + ) + self.assertEqual(runtime._session_directories[session.session_id], session_dir) + await runtime.close() + + async def test_audience_memory_rejects_a_tampered_route_before_launch(self) -> None: + runtime = PiriRuntime( + client_factory=self.factory, + process_environment={"BASE": "one"}, + memory_materializer_path="/materializer", + memory_environment_validator=lambda _value: (_ for _ in ()).throw( + ValueError("invalid route") + ), + ) + with self.assertRaisesRegex(ValueError, "invalid route"): + await runtime.start_or_resume( + SessionRequest( + working_directory="/workspace", + memory_environment={"UNTRUSTED": "path"}, + ) + ) + self.assertEqual(self.factory.clients, []) + + async def test_memory_materializer_without_route_validator_fails_closed(self) -> None: + runtime = PiriRuntime( + client_factory=self.factory, + process_environment={"BASE": "one"}, + memory_materializer_path="/materializer", + ) + with self.assertRaisesRegex(ValueError, "route validator"): + await runtime.start_or_resume( + SessionRequest( + working_directory="/workspace", + memory_environment={"UNTRUSTED": "path"}, + ) + ) + self.assertEqual(self.factory.clients, []) + async def test_resume_uses_and_verifies_exact_session_id(self) -> None: session = await self.runtime.start_or_resume( SessionRequest(working_directory="/workspace", session_id="piri-existing") @@ -439,6 +532,8 @@ async def test_auto_confirms_extension_yes_no_dialog(self) -> None: child = """ import json import sys +from pathlib import Path +Path("piri-child-state").write_text("private") command = json.loads(sys.stdin.readline()) request = { "type": "extension_ui_request", @@ -470,6 +565,13 @@ async def test_auto_confirms_extension_yes_no_dialog(self) -> None: await client.start() state = await client.get_state() self.assertEqual(state["sessionId"], "rpc-session") + if os.name == "posix": + self.assertEqual( + stat.S_IMODE( + (Path(directory) / "piri-child-state").stat().st_mode + ), + 0o600, + ) finally: await client.close() diff --git a/bridge/tests/test_piri_snapshot.py b/bridge/tests/test_piri_snapshot.py new file mode 100644 index 00000000..613e707b --- /dev/null +++ b/bridge/tests/test_piri_snapshot.py @@ -0,0 +1,105 @@ +"""Security and bounds tests for Piri transcript snapshots.""" + +from __future__ import annotations + +from datetime import datetime, timezone +import json +import os +from pathlib import Path + +import pytest + +from telegram_bot.memory.distill_types import TranscriptBounds +from telegram_bot.memory.piri_snapshot import read_piri_snapshot + + +def _write_session(directory: Path, session_id: str, entries: list[dict]) -> Path: + directory.mkdir(mode=0o700) + path = directory / f"2026-08-05T00-00-00-000Z_{session_id}.jsonl" + values = [ + { + "type": "session", + "version": 3, + "id": session_id, + "timestamp": "2026-08-05T00:00:00Z", + "cwd": "/workspace", + }, + *entries, + ] + path.write_text( + "".join(json.dumps(value) + "\n" for value in values), + encoding="utf-8", + ) + path.chmod(0o600) + return path + + +def test_reads_only_bounded_user_and_assistant_text(tmp_path: Path) -> None: + session_id = "piri-session" + session_dir = tmp_path / "sessions" + _write_session( + session_dir, + session_id, + [ + { + "type": "message", + "id": "m1", + "timestamp": "2026-08-05T00:00:01Z", + "message": {"role": "user", "content": "remember this"}, + }, + { + "type": "message", + "id": "tool", + "timestamp": "2026-08-05T00:00:02Z", + "message": {"role": "toolResult", "content": "raw tool output"}, + }, + { + "type": "message", + "id": "m2", + "timestamp": "2026-08-05T00:00:03Z", + "message": { + "role": "assistant", + "content": [{"type": "text", "text": "stored safely"}], + }, + }, + ], + ) + + snapshot = read_piri_snapshot( + session_dir, + session_id, + bounds=TranscriptBounds(max_messages=2, max_bytes=128), + now=datetime(2026, 8, 5, 0, 1, tzinfo=timezone.utc), + ) + + assert [(item.role, item.text) for item in snapshot.messages] == [ + ("user", "remember this"), + ("assistant", "stored safely"), + ] + assert snapshot.last_turn_id == "m2" + assert snapshot.byte_count == len("remember thisstored safely".encode()) + + +def test_rejects_identity_mismatch_and_symlinks(tmp_path: Path) -> None: + session_dir = tmp_path / "sessions" + path = _write_session(session_dir, "expected", []) + payload = path.read_text().replace('"id": "expected"', '"id": "other"', 1) + path.write_text(payload) + path.chmod(0o600) + with pytest.raises(ValueError, match="identity"): + read_piri_snapshot( + session_dir, + "expected", + bounds=TranscriptBounds(), + ) + + path.unlink() + target = tmp_path / "outside.jsonl" + target.write_text("{}\n") + os.symlink(target, path) + with pytest.raises(ValueError, match="unsafe"): + read_piri_snapshot( + session_dir, + "expected", + bounds=TranscriptBounds(), + ) diff --git a/bridge/tests/test_project_chat_codex.py b/bridge/tests/test_project_chat_codex.py index bc3fae76..811f0c39 100644 --- a/bridge/tests/test_project_chat_codex.py +++ b/bridge/tests/test_project_chat_codex.py @@ -298,6 +298,38 @@ async def test_project_chat_binds_opaque_claude_audiences_at_request_time( assert "-100456" not in serialized +@pytest.mark.anyio +async def test_project_chat_binds_opaque_piri_audiences_at_request_time( + tmp_path: Path, +) -> None: + settings = _settings(tmp_path, provider="piri") + settings.bridge_memory_mode = "audience-scoped" + settings.telegram_session_scope = "shared-groups" + settings.bot_data_dir = tmp_path / ".telegram_bot" + settings.bridge_memory_audience_root = None + settings.bridge_memory_audience_key_path = None + settings.bridge_unsafe_shared_all_memory = False + settings.honcho_memory_enabled = True + settings.honcho_config_path = tmp_path / ".hermes" / "honcho.json" + runtime = FakeRuntime() + handler = ProjectChatHandler(settings=settings, agent_runtime=runtime) + handler._task_ledger_cache = False + + await handler.process_message("private", 934719283, 934719283, new_session=True) + await handler.process_message("shared", 934719283, -100456, new_session=True) + + private_env = runtime.requests[0].memory_environment + shared_env = runtime.requests[1].memory_environment + assert private_env is not None and shared_env is not None + assert private_env["CCC_MEMORY_AUDIENCE"] == "private" + assert shared_env["CCC_MEMORY_AUDIENCE"] == "shared" + assert ( + private_env["PIRI_CODING_AGENT_SESSION_DIR"] + != shared_env["PIRI_CODING_AGENT_SESSION_DIR"] + ) + assert "CODEX_HOME" not in private_env + + def test_agent_provider_settings_default_and_reject_unknown(tmp_path: Path) -> None: settings_class = _real_settings_class() environ = {"HOME": str(tmp_path), "TELEGRAM_BOT_TOKEN": "123456:test"} diff --git a/bridge/tests/test_provider_capabilities.py b/bridge/tests/test_provider_capabilities.py index 2b54a423..88fb1b73 100644 --- a/bridge/tests/test_provider_capabilities.py +++ b/bridge/tests/test_provider_capabilities.py @@ -251,6 +251,34 @@ def test_codex_memory_parity_tracks_issue_465(self) -> None: (REPO_ROOT / "bridge/memory/distill_honcho_worker.py").is_file() ) + def test_piri_memory_parity_is_grounded_in_runtime_and_roundtrip_tests(self) -> None: + from telegram_bot.core.piri_runtime import PiriRuntime + + for axis in ( + "memory_read_bootstrap", + "memory_writeback_distill", + "memory_sink_local", + "memory_sink_honcho", + "memory_sink_wiki_candidate", + "memory_roundtrip", + ): + with self.subTest(axis=axis): + self.assertIs( + capability_status("piri", axis).state, + CapabilityState.SUPPORTED, + ) + self.assertIs( + capability_status("piri", "memory_postcompact_reinject").state, + CapabilityState.DEGRADED, + ) + self.assertTrue(hasattr(PiriRuntime, "read_session_snapshot")) + self.assertTrue( + (REPO_ROOT / "bridge/memory/piri_snapshot.py").is_file() + ) + self.assertTrue( + (REPO_ROOT / "bridge/tests/test_distill_roundtrip.py").is_file() + ) + def test_lifecycle_observability_claim_matches_the_opt_in_surfaces(self) -> None: from telegram_bot.core import ( lifecycle_audit, diff --git a/bridge/tests/test_session_composition.py b/bridge/tests/test_session_composition.py index 07efb65e..372286f5 100644 --- a/bridge/tests/test_session_composition.py +++ b/bridge/tests/test_session_composition.py @@ -1139,3 +1139,43 @@ def test_build_context_composes_routed_snapshot_worker(tmp_path): ) assert result.returncode == 0, result.stderr assert "COMPOSED-ROUTED-SNAPSHOT-WORKER-OK" in result.stdout + + +def test_build_context_composes_audience_scoped_piri_memory_pipeline(tmp_path): + result = _run_probe( + """ +import os +from pathlib import Path + +root = Path(os.environ["PROBE_ROOT"]) +(root / "project").mkdir(parents=True, exist_ok=True) +os.environ.update({ + "PROJECT_ROOT": str(root / "project"), + "TELEGRAM_BOT_TOKEN": "123456:test", + "ALLOWED_USER_IDS": "1", + "CCC_AGENT_PROVIDER": "piri", + "CCC_BRIDGE_MEMORY_MODE": "audience-scoped", + "CCC_BRIDGE_MEMORY_AUDIENCE_ROOT": str(root / "audiences"), + "CCC_HONCHO_MEMORY_ENABLED": "1", + "CCC_HONCHO_CFG": str(root / "honcho.json"), +}) + +from telegram_bot.__main__ import build_context, load_runtime_settings +from telegram_bot.core.piri_runtime import PiriRuntime +from telegram_bot.memory.codex_snapshot import CodexThreadSnapshotter + +context = build_context(load_runtime_settings()) +assert isinstance(context.agent_runtime, PiriRuntime) +assert context.agent_runtime._route_environment_factory is not None +assert context.agent_runtime._memory_environment_validator is not None +assert isinstance(context.distill_snapshot_worker, CodexThreadSnapshotter) +assert context.distill_snapshot_worker._runtime is context.agent_runtime +assert context.distill_local_sink_worker is not None +assert context.distill_wiki_sink_worker is not None +assert context.distill_honcho_sink_worker is not None +print("COMPOSED-PIRI-MEMORY-PIPELINE-OK") +""", + probe_root=tmp_path, + ) + assert result.returncode == 0, result.stderr + assert "COMPOSED-PIRI-MEMORY-PIPELINE-OK" in result.stdout diff --git a/bridge/tests/test_session_provider.py b/bridge/tests/test_session_provider.py index 7d7ec495..4456ea00 100644 --- a/bridge/tests/test_session_provider.py +++ b/bridge/tests/test_session_provider.py @@ -663,6 +663,32 @@ async def test_distill_command_records_current_codex_thread_without_reset( assert "remains active" in update.message.replies[0][0] +@pytest.mark.anyio +async def test_distill_command_records_current_piri_thread_without_reset( + tmp_path: Path, +) -> None: + manager = make_manager(tmp_path, "piri") + await manager.store.set( + "7:9", + { + "provider": "piri", + "session_id": "piri-current", + "last_user_message_at": "2026-08-05T02:00:00+00:00", + }, + ) + journal = RecordingDistillJournal() + bot = bare_bot(manager, provider="piri") + bot._distill_journal = journal + update = make_update(text="/distill") + + await bot._cmd_distill(update, SimpleNamespace(args=[])) + + assert journal.calls[0]["provider"] == "piri" + assert journal.calls[0]["thread_id"] == "piri-current" + assert "Piri memory distill request recorded" in update.message.replies[0][0] + assert (await manager.get_session("7:9"))["session_id"] == "piri-current" + + @pytest.mark.anyio async def test_distill_command_deduplicates_same_turn_and_allows_new_turn( tmp_path: Path, @@ -990,7 +1016,7 @@ async def test_auto_new_enqueues_old_codex_thread_before_reset(tmp_path: Path) - @pytest.mark.anyio -async def test_distill_trigger_is_noop_for_non_codex_or_missing_thread(tmp_path: Path) -> None: +async def test_distill_trigger_is_noop_for_unsupported_or_missing_thread(tmp_path: Path) -> None: manager = make_manager(tmp_path, "codex") bot = bare_bot(manager, provider="codex") journal = RecordingDistillJournal() @@ -1347,6 +1373,31 @@ async def test_checkpoint_turn_gate_enqueues_without_resetting_session( assert (await manager.get_session("7:9"))["session_id"] == "thread" +@pytest.mark.anyio +async def test_piri_checkpoint_turn_gate_enqueues_source_provider( + tmp_path: Path, +) -> None: + from telegram_bot.memory.distill_types import DistillTrigger + + manager = make_manager(tmp_path, "piri") + bot = bare_bot(manager, provider="piri") + journal = RecordingDistillJournal() + bot._distill_journal = journal + enable_checkpoint(bot, turns=1) + + await bot._save_session_id( + "7:9", + ChatResponse("assistant", session_id="piri-thread"), + user_id=7, + chat_id=9, + request_text="user", + turn_marker="message-1", + ) + + assert journal.calls[0]["provider"] == "piri" + assert journal.calls[0]["trigger"] is DistillTrigger.CHECKPOINT + + @pytest.mark.anyio async def test_checkpoint_uses_first_reached_utf8_byte_or_age_gate( tmp_path: Path, diff --git a/bridge/tests/test_shared_group_memory.py b/bridge/tests/test_shared_group_memory.py index 27c28b2e..78ecfc10 100644 --- a/bridge/tests/test_shared_group_memory.py +++ b/bridge/tests/test_shared_group_memory.py @@ -212,6 +212,32 @@ def test_audience_codex_environment_is_opaque_and_physically_separate( assert "-100456" not in serialized +def test_audience_piri_environment_is_opaque_and_physically_separate( + tmp_path: Path, +) -> None: + settings = _audience_settings(tmp_path) + private = resolve_memory_audience(settings, user_id=934719283, chat_id=934719283) + public = resolve_memory_audience(settings, user_id=934719283, chat_id=-100456) + assert private is not None and public is not None + + private_env = private.piri_environment(settings) + public_env = public.piri_environment(settings) + + assert private_env["PIRI_CODING_AGENT_SESSION_DIR"] == str( + private.scope_root / "piri" / "sessions" + ) + assert private_env["CCC_PIRI_BOOTSTRAP_CONTEXT_FILE"].endswith( + "/piri/bootstrap/AGENTS.md" + ) + assert ( + private_env["PIRI_CODING_AGENT_SESSION_DIR"] + != public_env["PIRI_CODING_AGENT_SESSION_DIR"] + ) + serialized = json.dumps((private_env, public_env), sort_keys=True) + assert "934719283" not in serialized + assert "-100456" not in serialized + + def test_bridge_memory_rejects_shared_all_without_unsafe_legacy_override( tmp_path: Path, ) -> None: diff --git a/bridge/utils/memory_policy.py b/bridge/utils/memory_policy.py index 9c4c3497..1783c6d2 100644 --- a/bridge/utils/memory_policy.py +++ b/bridge/utils/memory_policy.py @@ -46,13 +46,6 @@ def assert_memory_scope_safe( "audience homes." ) -_PIRI_AUDIENCE_ERROR = ( - "audience-scoped memory cannot run with the Piri provider: PiriRuntime " - "does not yet isolate configuration, credentials, and session storage per " - "memory audience. Use CCC_BRIDGE_MEMORY_MODE=off or curated." -) - - def assert_memory_provider_safe( mode: str, provider: str, @@ -62,7 +55,9 @@ def assert_memory_provider_safe( Codex is allowed only with its officially supported OS keyring credential store. A file-backed login lives inside ``CODEX_HOME`` and copying it into - each audience would multiply long-lived access tokens. + each audience would multiply long-lived access tokens. Piri keeps its + operator-owned provider configuration global while the runtime separately + isolates transcripts and generated memory context per audience. """ if ( @@ -71,8 +66,3 @@ def assert_memory_provider_safe( and str(codex_audience_auth_mode or "").strip().lower() != "keyring" ): raise ValueError(_CODEX_AUDIENCE_ERROR) - if ( - mode == MEMORY_MODE_AUDIENCE_SCOPED - and str(provider or "").strip().lower() == "piri" - ): - raise ValueError(_PIRI_AUDIENCE_ERROR) diff --git a/claude/hooks/nunchi/piri-feed.sh b/claude/hooks/nunchi/piri-feed.sh index ead3c10e..be9c19e1 100644 --- a/claude/hooks/nunchi/piri-feed.sh +++ b/claude/hooks/nunchi/piri-feed.sh @@ -1,12 +1,11 @@ #!/usr/bin/env bash # nunchi piri-feed extractor (#816) — Piri-provider nodes. # -# Piri sessions are not read by the Claude/Codex distill journal (the Piri RPC -# runtime exposes no distill/write-back extractor), so — like codex-feed.sh on a -# Codex node — this lane extracts user/agent messages from NEW Piri session -# jsonl files, asks the configured Piri CLI for distill-style facts in one -# non-interactive print-mode run, and ingests them into the nunchi peer_facts -# DB. Idempotent via a seen-file; bounded per run. Runs from cron. +# This supplementary nunchi lane extracts user/agent messages from NEW Piri +# session JSONL files, asks the configured Piri CLI for peer facts in one +# non-interactive print-mode run, and ingests them into the nunchi DB. The main +# ccc distill journal independently owns local/Honcho/Wiki write-back. +# Idempotent via a seen-file; bounded per run. Runs from cron. # NOTE: unlike ingest-cron.sh this costs one Piri run per new file. # No-op unless nunchi is enabled (state/nunchi.mode=on or CCC_NUNCHI_MODE=on). set -uo pipefail @@ -20,8 +19,15 @@ FM="$HERE/nunchi.py" NUNCHI_HOME="${NUNCHI_HOME:-$HOME/.nunchi}" SEEN="$NUNCHI_HOME/piri-seen" LOCK="$NUNCHI_HOME/.piri-feed.lock" -# Piri stores sessions under /sessions//.jsonl. -PIR_SESSIONS_DIR="${PIR_SESSIONS_DIR:-${PIRI_CODING_AGENT_SESSION_DIR:-$HOME/.piri/agent}/sessions}" +# PIRI_CODING_AGENT_SESSION_DIR is already the direct session directory. The +# global default is /sessions and may contain cwd subdirectories. +if [ -z "${PIR_SESSIONS_DIR:-}" ]; then + if [ -n "${PIRI_CODING_AGENT_SESSION_DIR:-}" ]; then + PIR_SESSIONS_DIR="$PIRI_CODING_AGENT_SESSION_DIR" + else + PIR_SESSIONS_DIR="${PIRI_CODING_AGENT_DIR:-$HOME/.piri/agent}/sessions" + fi +fi # The Piri CLI the bridge runs (ccc-node PiriRuntime). Falls back to `piri` on PATH. PIR_CLI="${CCC_PIRI_CLI_PATH:-piri}" # Isolate extractor runs so their own sessions never land under PIR_SESSIONS_DIR diff --git a/docs/piri-runtime-contract.md b/docs/piri-runtime-contract.md index 10646551..62a15a5f 100644 --- a/docs/piri-runtime-contract.md +++ b/docs/piri-runtime-contract.md @@ -45,6 +45,25 @@ user that runs ccc-node. not expose a bounded stored-session browser, so `/resume` cannot list or preview arbitrary Piri sessions. +## Memory contract + +- Audience-scoped memory gives every audience private Piri transcript and + generated-bootstrap directories. Piri's global provider configuration and + credential store remain shared and are not copied into those directories. +- ccc-node disables Piri's automatic project context discovery with + `--no-context-files` and passes only the audience-scoped generated + `AGENTS.md` as an appended system prompt. +- The stored-session reader accepts only bounded, owner-only, non-symlink + Piri JSONL transcripts whose header contains the exact requested session id. +- Session-end, provider-switch, `/new`, shutdown, and `/distill` checkpoints + enter the same local, Honcho, and Family Wiki writeback pipeline used by + Codex. The source provenance remains `piri`. +- Writeback extraction currently uses the isolated Codex extraction backend, + so a working Codex CLI and authentication are required for Piri distillation; + the interactive Piri turn still runs on its configured Kimi or GLM model. +- Piri RPC 0.83 has no post-compaction hook. Read bootstrap and session-end + writeback are supported, while post-compaction refresh remains degraded. + ## Telegram surface - `/model` uses Piri's live `get_available_models` catalog. @@ -56,9 +75,9 @@ user that runs ccc-node. expose normalized tokens, account quota, or reset windows. - Startup readiness requires both a working Piri CLI and at least one model returned by RPC discovery for the configured Piri auth store. -- Audience-scoped ccc memory is rejected for Piri until configuration, - credentials, and session storage can be isolated per audience. `off` and - `curated` memory modes remain available. +- Audience-scoped ccc memory is supported when the exact generated-bootstrap + and session-storage routes pass fail-closed validation. `off` and `curated` + memory modes remain available. ## Event contract diff --git a/docs/provider-capability-matrix.md b/docs/provider-capability-matrix.md index 50657e1d..caafb159 100644 --- a/docs/provider-capability-matrix.md +++ b/docs/provider-capability-matrix.md @@ -58,10 +58,10 @@ column here before landing. | Capability | claude | codex | crush | piri | |---|---|---|---|---| | `memory_session_resume` — Memory: session resume: Conversation context resumes from the provider thread/session id. | `supported` — Persisted SDK session ids resume with full provider-side context. | `supported` — Thread ids persist per conversation and resume through thread/resume; live cold resume verified 2026-07-15. | `supported` — Sessions resume by id with provider-side context; the crush server persists session history in its data dir. | `supported` — Piri session ids persist per Telegram conversation and resume through --session-id with exact get_state verification. | -| `memory_read_bootstrap` — Memory: read bootstrap: The MEMORY/USER/local/Wiki/Honcho/resume startup snapshot is recognized at session start. | `supported` — SessionStart injects the bounded local snapshot via claude/hooks/load-memory.sh. | `supported` — The AGENTS.override.md materializer runs before thread start/resume; promoted after the 2026-07-15 live gate (#419). | `unsupported` — No startup-snapshot materializer is wired for crush sessions. (depends on #926) | `degraded` — Piri starts in the project directory and can consume project context, but the ccc audience-scoped memory materializer is intentionally disabled. | -| `memory_postcompact_reinject` — Memory: post-compaction reinjection: Instruction/memory meaning survives compaction and cold resume. | `supported` — PostCompact re-injects the bounded snapshot through claude/hooks/load-memory.sh. | `degraded` — Cold resume re-reads the refreshed global snapshot, but the app-server exposes no official PreCompact/PostCompact event and turn/completed is not treated as compaction. Provider compaction checkpoint/reinjection therefore remains unverified. | `degraded` — crush exposes no compaction lifecycle event to hook. (depends on #926) | `unsupported` — ccc-node has no Piri compaction checkpoint or post-compaction memory reinjection hook. | -| `memory_writeback_distill` — Memory: write-back distill: Durable facts are extracted from provider threads for the memory sinks. | `supported` — PreCompact/SessionEnd distill transcripts into resume/local facts and sink candidates via claude/hooks/distill.sh. | `supported` — Session-reset, explicit, opt-in bounded checkpoint, and bounded shutdown-queue triggers, extraction, and the local sink are scheduled. Provider/model/turn-byte/duration accounting and shared warn/enforce cost gates are body-free; Wiki candidates enter a local human-review queue and Honcho facts use an owner-only retrying outbox. | `unsupported` — No distill snapshotter is wired for crush sessions. (depends on #926) | `unsupported` — Piri sessions are not read by the Codex/Claude distill journal, so the local/Honcho/Wiki distill sinks stay unwired. (A provider-neutral nunchi peer-facts extractor, `hooks/nunchi/piri-feed.sh` via `install-nunchi.sh --piri`, does feed the nunchi DB from Piri sessions.) | -| `memory_sink_local` — Memory: local sink: resume/local structured facts are written replay-safe with provenance. | `supported` — Distill writes resume.md and local structured facts (claude/hooks/distill/resume-write.sh, local-facts.sh). | `supported` — Supported session-reset triggers bind an opaque audience route, and an independently leased worker writes replay-safe local facts/resume. | `unsupported` — Local memory sink is not wired for crush. (depends on #926) | `unsupported` — No Piri write-back extractor feeds the replay-safe local memory sink. | -| `memory_sink_honcho` — Memory: Honcho sink: Redacted conclusions push to Honcho through a durable retry queue. | `supported` — Redacted payloads push via claude/hooks/distill/honcho-push.sh with the queue-drain retry path. | `supported` — Validated facts use an owner-only per-job outbox, stable idempotency keys, independently leased retries, and body-free failures. Audience-scoped jobs additionally use scope-partitioned outboxes and physically distinct Honcho workspaces; unscoped jobs fail closed in that mode. | `unsupported` — Honcho sink is not wired for crush. (depends on #926) | `unsupported` — No Piri write-back extractor feeds the Honcho outbox. | -| `memory_sink_wiki_candidate` — Memory: Wiki candidate sink: Only human-gated Wiki candidates are generated; nothing auto-merges. | `supported` — Distill emits human-gated Wiki candidates via claude/hooks/distill/wiki-queue.sh. | `supported` — Validated candidates are atomically queued in owner-only per-job records; the sink performs no Wiki write, branch, PR, or merge. | `unsupported` — Wiki-candidate sink is not wired for crush. (depends on #926) | `unsupported` — No Piri write-back extractor feeds the human-gated Wiki candidate queue. | -| `memory_roundtrip` — Memory: read/write round-trip: A durable fact written in session A is recalled by an isolated later session B. | `supported` — SessionEnd write-back feeds the next SessionStart snapshot; both hook directions carry executable tests (memory-hooks.test.sh, distill/*.test.sh). | `supported` — The hermetic audience-scoped test and an approved isolated live provider A→distill→local index→B run on 2026-07-23 both recalled one durable fact exactly once; local, Wiki-candidate, and Honcho sink states remain independently replayable (#465). | `unsupported` — Depends on the distill/writeback chain, which crush does not wire yet. (depends on #926) | `unsupported` — A Piri session A to durable write-back to isolated session B round-trip does not exist because Piri write-back is not implemented. | +| `memory_read_bootstrap` — Memory: read bootstrap: The MEMORY/USER/local/Wiki/Honcho/resume startup snapshot is recognized at session start. | `supported` — SessionStart injects the bounded local snapshot via claude/hooks/load-memory.sh. | `supported` — The AGENTS.override.md materializer runs before thread start/resume; promoted after the 2026-07-15 live gate (#419). | `unsupported` — No startup-snapshot materializer is wired for crush sessions. (depends on #926) | `supported` — Audience-scoped Piri sessions use isolated transcript directories, disable automatic AGENTS/CLAUDE discovery, run the bounded ccc materializer, and append only the generated scope-local AGENTS.md context. | +| `memory_postcompact_reinject` — Memory: post-compaction reinjection: Instruction/memory meaning survives compaction and cold resume. | `supported` — PostCompact re-injects the bounded snapshot through claude/hooks/load-memory.sh. | `degraded` — Cold resume re-reads the refreshed global snapshot, but the app-server exposes no official PreCompact/PostCompact event and turn/completed is not treated as compaction. Provider compaction checkpoint/reinjection therefore remains unverified. | `degraded` — crush exposes no compaction lifecycle event to hook. (depends on #926) | `degraded` — Cold start/resume refreshes and re-appends the scoped snapshot, but Piri exposes no ccc compaction lifecycle hook for an explicit mid-session checkpoint or reinjection proof. | +| `memory_writeback_distill` — Memory: write-back distill: Durable facts are extracted from provider threads for the memory sinks. | `supported` — PreCompact/SessionEnd distill transcripts into resume/local facts and sink candidates via claude/hooks/distill.sh. | `supported` — Session-reset, explicit, opt-in bounded checkpoint, and bounded shutdown-queue triggers, extraction, and the local sink are scheduled. Provider/model/turn-byte/duration accounting and shared warn/enforce cost gates are body-free; Wiki candidates enter a local human-review queue and Honcho facts use an owner-only retrying outbox. | `unsupported` — No distill snapshotter is wired for crush sessions. (depends on #926) | `supported` — Session-reset, explicit, checkpoint, and shutdown triggers enter the provider-neutral journal. A secure bounded Piri JSONL snapshot preserves source provider provenance through the isolated extractor and all sinks. | +| `memory_sink_local` — Memory: local sink: resume/local structured facts are written replay-safe with provenance. | `supported` — Distill writes resume.md and local structured facts (claude/hooks/distill/resume-write.sh, local-facts.sh). | `supported` — Supported session-reset triggers bind an opaque audience route, and an independently leased worker writes replay-safe local facts/resume. | `unsupported` — Local memory sink is not wired for crush. (depends on #926) | `supported` — Audience-routed Piri jobs use the same independently leased replay-safe local facts/resume sink while retaining provider=piri provenance. | +| `memory_sink_honcho` — Memory: Honcho sink: Redacted conclusions push to Honcho through a durable retry queue. | `supported` — Redacted payloads push via claude/hooks/distill/honcho-push.sh with the queue-drain retry path. | `supported` — Validated facts use an owner-only per-job outbox, stable idempotency keys, independently leased retries, and body-free failures. Audience-scoped jobs additionally use scope-partitioned outboxes and physically distinct Honcho workspaces; unscoped jobs fail closed in that mode. | `unsupported` — Honcho sink is not wired for crush. (depends on #926) | `supported` — Validated Piri facts use the same owner-only scope-partitioned Honcho outbox, idempotency keys, retry leases, and distinct workspaces. | +| `memory_sink_wiki_candidate` — Memory: Wiki candidate sink: Only human-gated Wiki candidates are generated; nothing auto-merges. | `supported` — Distill emits human-gated Wiki candidates via claude/hooks/distill/wiki-queue.sh. | `supported` — Validated candidates are atomically queued in owner-only per-job records; the sink performs no Wiki write, branch, PR, or merge. | `unsupported` — Wiki-candidate sink is not wired for crush. (depends on #926) | `supported` — Validated Piri candidates enter the same owner-only human-review queue; the sink still performs no Wiki write, branch, PR, or merge. | +| `memory_roundtrip` — Memory: read/write round-trip: A durable fact written in session A is recalled by an isolated later session B. | `supported` — SessionEnd write-back feeds the next SessionStart snapshot; both hook directions carry executable tests (memory-hooks.test.sh, distill/*.test.sh). | `supported` — The hermetic audience-scoped test and an approved isolated live provider A→distill→local index→B run on 2026-07-23 both recalled one durable fact exactly once; local, Wiki-candidate, and Honcho sink states remain independently replayable (#465). | `unsupported` — Depends on the distill/writeback chain, which crush does not wire yet. (depends on #926) | `supported` — The hermetic audience-scoped Piri A→snapshot→distill→local index→B bootstrap test recalls one durable fact with provider=piri provenance. | diff --git a/schemas/codex-distill-extraction-v1.schema.json b/schemas/codex-distill-extraction-v1.schema.json index 6e5a402c..88b2c458 100644 --- a/schemas/codex-distill-extraction-v1.schema.json +++ b/schemas/codex-distill-extraction-v1.schema.json @@ -9,7 +9,10 @@ "type": "string" }, "provider": { - "const": "codex", + "enum": [ + "codex", + "piri" + ], "title": "Provider", "type": "string" }, diff --git a/scripts/ccc_codex_memory.py b/scripts/ccc_codex_memory.py index 604e474e..6962882d 100755 --- a/scripts/ccc_codex_memory.py +++ b/scripts/ccc_codex_memory.py @@ -967,8 +967,6 @@ def _audience_scoped_blocked(environ: Mapping[str, str] | None = None) -> bool: raw = (env.get("CCC_MEMORY_AUDIENCE_SCOPED") or "").strip().lower() if raw in ("", "0", "false", "off", "no"): return False - if (env.get("CCC_CODEX_AUDIENCE_AUTH_MODE") or "").strip().lower() != "keyring": - return True audience = (env.get("CCC_MEMORY_AUDIENCE") or "").strip().lower() scope = (env.get("CCC_MEMORY_SCOPE") or "").strip() if audience == "shared": @@ -980,6 +978,7 @@ def _audience_scoped_blocked(environ: Mapping[str, str] | None = None) -> bool: else: return True root_raw = (env.get("CCC_MEMORY_AUDIENCE_ROOT") or "").strip() + provider = (env.get("CCC_MEMORY_MATERIALIZER_PROVIDER") or "codex").strip().lower() codex_raw = (env.get("CODEX_HOME") or "").strip() sqlite_raw = (env.get("CODEX_SQLITE_HOME") or "").strip() if not root_raw or not codex_raw or not sqlite_raw: @@ -989,6 +988,35 @@ def _audience_scoped_blocked(environ: Mapping[str, str] | None = None) -> bool: sqlite_home = Path(sqlite_raw).expanduser() if not root.is_absolute() or not codex_home.is_absolute() or not sqlite_home.is_absolute(): return True + if provider == "piri": + bootstrap_raw = (env.get("CCC_PIRI_BOOTSTRAP_HOME") or "").strip() + session_raw = (env.get("PIRI_CODING_AGENT_SESSION_DIR") or "").strip() + context_raw = (env.get("CCC_PIRI_BOOTSTRAP_CONTEXT_FILE") or "").strip() + if not bootstrap_raw or not session_raw or not context_raw: + return True + bootstrap_home = Path(bootstrap_raw).expanduser() + session_dir = Path(session_raw).expanduser() + context_file = Path(context_raw).expanduser() + if ( + not bootstrap_home.is_absolute() + or not session_dir.is_absolute() + or not context_file.is_absolute() + ): + return True + expected = Path(os.path.abspath(root / scope / "piri" / "bootstrap")) + expected_session = Path(os.path.abspath(root / scope / "piri" / "sessions")) + expected_context = expected / "AGENTS.md" + return ( + Path(os.path.abspath(codex_home)) != expected + or Path(os.path.abspath(sqlite_home)) != expected + or Path(os.path.abspath(bootstrap_home)) != expected + or Path(os.path.abspath(session_dir)) != expected_session + or Path(os.path.abspath(context_file)) != expected_context + ) + if provider != "codex": + return True + if (env.get("CCC_CODEX_AUDIENCE_AUTH_MODE") or "").strip().lower() != "keyring": + return True expected = Path(os.path.abspath(root / scope / "codex")) return ( Path(os.path.abspath(codex_home)) != expected diff --git a/scripts/ccc_codex_memory_test.py b/scripts/ccc_codex_memory_test.py index ff813f88..681cfd44 100644 --- a/scripts/ccc_codex_memory_test.py +++ b/scripts/ccc_codex_memory_test.py @@ -592,6 +592,29 @@ def test_audience_scoped_off_spellings_do_not_block(self) -> None: {**shared, "CCC_MEMORY_SCOPE": "../private-leak"} ) ) + piri_scope = "private-" + "b" * 32 + piri_root = self.root / "piri-audiences" + piri_home = piri_root / piri_scope / "piri" + piri = { + "CCC_MEMORY_AUDIENCE_SCOPED": "1", + "CCC_MEMORY_AUDIENCE": "private", + "CCC_MEMORY_AUDIENCE_ROOT": str(piri_root), + "CCC_MEMORY_SCOPE": piri_scope, + "CCC_MEMORY_MATERIALIZER_PROVIDER": "piri", + "CCC_PIRI_BOOTSTRAP_HOME": str(piri_home / "bootstrap"), + "PIRI_CODING_AGENT_SESSION_DIR": str(piri_home / "sessions"), + "CCC_PIRI_BOOTSTRAP_CONTEXT_FILE": str( + piri_home / "bootstrap" / "AGENTS.md" + ), + "CODEX_HOME": str(piri_home / "bootstrap"), + "CODEX_SQLITE_HOME": str(piri_home / "bootstrap"), + } + self.assertFalse(self.module._audience_scoped_blocked(piri)) + self.assertTrue( + self.module._audience_scoped_blocked( + {**piri, "PIRI_CODING_AGENT_SESSION_DIR": str(self.root / "leak")} + ) + ) def test_loader_and_errors_are_bounded_body_free_codes(self) -> None: with self.assertRaises(self.module.MaterializeError) as caught: diff --git a/scripts/ccc_doctor.py b/scripts/ccc_doctor.py index c5c05d7b..0478a92e 100644 --- a/scripts/ccc_doctor.py +++ b/scripts/ccc_doctor.py @@ -99,6 +99,7 @@ def __init__(self, repo: Path, claude_dir: Path, scope: str): self.settings_valid = False self.current_settings: dict[str, Any] | None = None self._rewrite_pairs: dict[str, str] | None = None + self._bridge_provider_state: tuple[str, str] | None = None def add(self, klass: str, item: str, status: str, action: str) -> None: self.rows.append(Row(klass, item, status, action)) @@ -457,13 +458,26 @@ def check_codex_managed_skills(self) -> None: def check_provider_readiness(self) -> None: if self.provider == "claude": return + if self.provider == "piri": + if self._bridge_provider_state == ("piri", "healthy"): + self.readiness = "ready" + self.add("정상", "Piri runtime", "healthy", "none") + else: + self.readiness = "failed" + self.add( + "수동필요", + "Piri runtime", + "live readiness not proven", + "inspect bridge status and Piri provider authentication", + ) + return if self.provider != "codex": self.readiness = "failed" self.add( "수동필요", "agent provider", "unsupported provider", - "set CCC_AGENT_PROVIDER to claude or codex", + "set CCC_AGENT_PROVIDER to claude, codex, or piri", ) return @@ -644,6 +658,26 @@ def bridge_status_verdict(returncode: int, output: str) -> tuple[str, str]: return "경고", "unavailable" return "경고", "unrecognized status output" if output.strip() else "no status output" + @staticmethod + def bridge_status_provider(output: str) -> tuple[str, str] | None: + """Return the single body-free provider label rendered by start.sh.""" + + labels = { + "Claude": "claude", + "Codex": "codex", + "Piri": "piri", + } + found: list[tuple[str, str]] = [] + for label, provider in labels.items(): + match = re.search( + rf"^\s*{label}:\s+(healthy|degraded|unavailable)\b", + output, + re.MULTILINE, + ) + if match: + found.append((provider, match.group(1))) + return found[0] if len(found) == 1 else None + def check_bridge_status(self) -> None: start = self.repo / "bridge/start.sh" if os.access(start, os.X_OK): @@ -656,6 +690,10 @@ def check_bridge_status(self) -> None: try: out = subprocess.run(["bash", str(start), "--path", probe_home, "--status"], text=True, capture_output=True, timeout=20) output = out.stdout + out.stderr + detected_provider = self.bridge_status_provider(output) + if detected_provider is not None: + self._bridge_provider_state = detected_provider + self.provider = detected_provider[0] klass, status = self.bridge_status_verdict(out.returncode, output) action = "none" if klass == "정상" else "inspect bridge service and body-free health diagnostics" self.add(klass, "bridge status", status, action) diff --git a/scripts/ccc_doctor_bridge_status_test.py b/scripts/ccc_doctor_bridge_status_test.py index fe05f8c3..a8591fe9 100644 --- a/scripts/ccc_doctor_bridge_status_test.py +++ b/scripts/ccc_doctor_bridge_status_test.py @@ -35,6 +35,27 @@ def test_nonzero_or_unrecognized_output_is_warning(self) -> None: self.assertEqual(Doctor.bridge_status_verdict(3, "some output")[0], "경고") self.assertEqual(Doctor.bridge_status_verdict(0, "some output")[0], "경고") + def test_provider_parser_identifies_one_live_piri_lane(self) -> None: + output = "🟢 Bot status: available\n Telegram: healthy\n Piri: healthy\n" + self.assertEqual(Doctor.bridge_status_provider(output), ("piri", "healthy")) + + def test_provider_parser_rejects_ambiguous_or_missing_labels(self) -> None: + self.assertIsNone(Doctor.bridge_status_provider("Bot status: available")) + self.assertIsNone( + Doctor.bridge_status_provider("Codex: healthy\nPiri: healthy\n") + ) + + def test_healthy_live_piri_status_is_ready(self) -> None: + doctor = Doctor(Path.cwd(), Path.cwd() / ".claude", "settings") + doctor.provider = "piri" + doctor._bridge_provider_state = ("piri", "healthy") + + doctor.check_provider_readiness() + + self.assertEqual(doctor.readiness, "ready") + self.assertEqual(doctor.rows[-1].item, "Piri runtime") + self.assertEqual(doctor.rows[-1].klass, "정상") + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/scripts/install-nunchi.sh b/scripts/install-nunchi.sh index 5f47d1cb..9fdecb18 100755 --- a/scripts/install-nunchi.sh +++ b/scripts/install-nunchi.sh @@ -12,8 +12,9 @@ # install-nunchi.sh # status # # Claude retains the standalone SessionStart hook and reuses the Session -# Distiller output (zero LLM cost). Codex and Piri have no distill feed, so -# their lanes run a per-new-session extractor (codex exec / Piri print mode). +# Distiller output (zero LLM cost). Codex and Piri keep supplementary nunchi +# per-new-session extractors (codex exec / Piri print mode); the main bridge +# distill journal separately owns replay-safe memory sinks. # Provider changes remove the other path so one runtime never injects the same # node-global snapshot twice. set -euo pipefail From f79f8c109405c2183faf06deeed2ca458ae0f5a9 Mon Sep 17 00:00:00 2001 From: jinon86 <247078695+jinon86@users.noreply.github.com> Date: Wed, 5 Aug 2026 12:34:38 +0900 Subject: [PATCH 2/3] refactor: reduce bridge composition complexity --- bridge/__main__.py | 124 +++++++++++++++++++++++++-------------------- 1 file changed, 68 insertions(+), 56 deletions(-) diff --git a/bridge/__main__.py b/bridge/__main__.py index 7bbd6387..596268d5 100644 --- a/bridge/__main__.py +++ b/bridge/__main__.py @@ -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,47 +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.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") - agent_runtime = 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, - ) + agent_runtime = _build_piri_runtime(settings) telegram_port = telegram_port or Application.builder clock = clock or time bind_logs_dir(settings.logs_dir) @@ -227,21 +253,7 @@ def validate_piri_memory_environment(environment: Mapping[str, 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( From 2b92eaee56fdbeb3cec33b86eebfc7baab6988b6 Mon Sep 17 00:00:00 2001 From: jinon86 <247078695+jinon86@users.noreply.github.com> Date: Wed, 5 Aug 2026 12:39:30 +0900 Subject: [PATCH 3/3] fix: narrow distill source provider type --- bridge/memory/distill_worker.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/bridge/memory/distill_worker.py b/bridge/memory/distill_worker.py index 1e5ab915..2c70952f 100644 --- a/bridge/memory/distill_worker.py +++ b/bridge/memory/distill_worker.py @@ -9,7 +9,7 @@ import re import secrets import time -from typing import Protocol +from typing import Literal, Protocol, cast from .distill_extraction import ( DistillBackend, @@ -260,7 +260,7 @@ async def extract_once(self, *, job_id: str) -> DistillJob: extraction_input = build_extraction_input( snapshot, trigger=claimed.trigger, - provider=claimed.provider, + provider=cast(Literal["codex", "piri"], claimed.provider), ) except (TypeError, ValueError): self._refund_unused_reservation(reservation)