From 726c6cd87ba0f87fc9eabcb03ffd2579a250c89c Mon Sep 17 00:00:00 2001 From: jinon86 <247078695+jinon86@users.noreply.github.com> Date: Thu, 6 Aug 2026 09:50:35 +0000 Subject: [PATCH] feat(memory): kick scoped nunchi ingest on distill local-sink completion On audience-scoped Piri nodes the nunchi lane (primary working memory during the gate-3 transition) only ingested via the 10-minute cron dispatcher, so a session started right after the previous one closed was injected with pre-close memory. The distill local-sink worker now fires one detached, body-free scoped piri-feed run for the job's own route once the sink+index land: the closed session reaches the scope DB/snapshot before the next session starts. The cron dispatcher remains the owner of record; feed-side flock + seen-file keep overlap idempotent, kick failures never affect job state, and an absent/unsafe feed path skips silently. --- CHANGELOG.md | 7 ++ bridge/__main__.py | 7 ++ bridge/memory/distill_local_worker.py | 136 ++++++++++++++++++++++ bridge/tests/test_distill_local_worker.py | 130 +++++++++++++++++++++ 4 files changed, 280 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index fd5c54f5..5a84d89a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ All notable changes to the Claude Code node harness. Dates are KST. ## [Unreleased] ### Added +- Session-close nunchi kick for audience-scoped Piri nodes. When a distill + local-sink job lands, the bridge now fires one detached, body-free scoped + `piri-feed.sh` run for that route, so the just-closed session reaches the + scope's nunchi DB/snapshot before the next session starts instead of + waiting for the 10-minute cron. The cron dispatcher stays the owner of + record; feed-side flock and seen-file keep overlap idempotent, the kick + never changes job state, and an absent/unsafe feed path skips silently. - Audience-scoped Piri/Nunchi/MemPalace collection and recall (#950). The bridge now supplies one canonical Nunchi DB/snapshot and isolated MemPalace HOME per opaque memory audience. Private recall is private + shared + diff --git a/bridge/__main__.py b/bridge/__main__.py index fbb1d3e3..0cb40228 100644 --- a/bridge/__main__.py +++ b/bridge/__main__.py @@ -309,6 +309,13 @@ def route_environment(audience: str, scope: str): Path(settings.codex_memory_materializer_path).expanduser().parent / "ccc-memory-index.sh" ), + nunchi_feed_path=( + Path(settings.codex_memory_materializer_path).expanduser().parent + / "nunchi" + / "piri-feed.sh" + if settings.agent_provider == "piri" + else None + ), ) from telegram_bot.memory.promotion import CodexMemoryPromoter diff --git a/bridge/memory/distill_local_worker.py b/bridge/memory/distill_local_worker.py index 8255e02a..8e5a3430 100644 --- a/bridge/memory/distill_local_worker.py +++ b/bridge/memory/distill_local_worker.py @@ -4,6 +4,7 @@ import asyncio from collections.abc import Mapping +from contextlib import suppress import math import os from pathlib import Path @@ -26,7 +27,14 @@ "LC_ALL", "LC_CTYPE", ) +# Passthrough for the session-close nunchi kick: the feed prefers the bridge's +# Piri wrapper (which itself skips memory bootstrap for extractor sessions). +_NUNCHI_FEED_ENV_PASSTHROUGH = ( + "CCC_PIRI_CLI_PATH", + "CCC_PIRI_REAL_CLI_PATH", +) _MAX_INDEXER_BYTES = 1024 * 1024 +_MAX_NUNCHI_FEED_BYTES = 256 * 1024 class _LocalIndexError(RuntimeError): @@ -80,6 +88,8 @@ def __init__( max_resume_bytes: int = 4000, indexer_path: str | Path | None = None, index_timeout_seconds: float = 30.0, + nunchi_feed_path: str | Path | None = None, + nunchi_feed_timeout_seconds: float = 900.0, environment: Mapping[str, str] | None = None, ) -> None: if lease_seconds <= 0 or max_attempts <= 0: @@ -94,6 +104,14 @@ def __init__( or index_timeout_seconds > 60 ): raise ValueError("invalid local sink index timeout") + if ( + not isinstance(nunchi_feed_timeout_seconds, (int, float)) + or isinstance(nunchi_feed_timeout_seconds, bool) + or not math.isfinite(nunchi_feed_timeout_seconds) + or nunchi_feed_timeout_seconds <= 0 + or nunchi_feed_timeout_seconds > 3600 + ): + raise ValueError("invalid nunchi feed kick timeout") self._journal = journal self._audience_root = Path(os.path.abspath(os.fspath(audience_root))) self._owner_token = owner_token or secrets.token_hex(16) @@ -103,6 +121,11 @@ def __init__( self._max_resume_bytes = max_resume_bytes self._indexer_path = Path(indexer_path) if indexer_path is not None else None self._index_timeout_seconds = float(index_timeout_seconds) + self._nunchi_feed_path = ( + Path(nunchi_feed_path) if nunchi_feed_path is not None else None + ) + self._nunchi_feed_timeout_seconds = float(nunchi_feed_timeout_seconds) + self._nunchi_tasks: set[asyncio.Task[None]] = set() self._environment = dict(os.environ if environment is None else environment) async def _fail( @@ -165,6 +188,118 @@ def _validated_indexer(self) -> Path | None: raise _LocalIndexError(terminal=True) return candidate + def _validated_nunchi_feed(self) -> Path | None: + """Best-effort nunchi feed path: any doubt means skip (cron is the fallback).""" + + if self._nunchi_feed_path is None: + return None + candidate = Path(os.path.abspath(os.fspath(self._nunchi_feed_path))) + try: + metadata = candidate.lstat() + except OSError: + return None + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or metadata.st_nlink != 1 + or metadata.st_uid not in {0, os.geteuid()} + or stat.S_IMODE(metadata.st_mode) & 0o022 + or metadata.st_size <= 0 + or metadata.st_size > _MAX_NUNCHI_FEED_BYTES + or not os.access(candidate, os.X_OK) + ): + return None + return candidate + + def _nunchi_feed_environment( + self, + *, + audience: str, + scope: str, + ) -> dict[str, str]: + try: + validate_memory_route(audience, scope) + except ValueError: + raise _LocalIndexError(terminal=True) + scope_root = self._audience_root / scope + nunchi_home = scope_root / "nunchi" + sessions_dir = scope_root / "piri" / "sessions" + environment = { + name: value + for name in _INDEX_ENV_ALLOWLIST + _NUNCHI_FEED_ENV_PASSTHROUGH + if isinstance((value := self._environment.get(name)), str) + and "\x00" not in value + } + environment.setdefault("PATH", "/usr/local/bin:/usr/bin:/bin") + environment.update( + { + # Same shape the cron dispatcher uses for one scoped child run. + "CCC_NUNCHI_SCOPED_CHILD": "1", + "CCC_NUNCHI_AUDIENCE_SCOPE": scope, + "CCC_NUNCHI_AUDIENCE_KIND": audience, + "PIRI_CODING_AGENT_SESSION_DIR": str(sessions_dir), + "PIR_SESSIONS_DIR": str(sessions_dir), + "NUNCHI_HOME": str(nunchi_home), + "NUNCHI_DB": str(nunchi_home / "facts.db"), + "NUNCHI_SNAPSHOT": str(nunchi_home / "snapshot.md"), + "PYTHONDONTWRITEBYTECODE": "1", + } + ) + return environment + + async def _drain_nunchi_feed(self, process: asyncio.subprocess.Process) -> None: + """Reap one detached kick; failures never reach the job (cron catches up).""" + + try: + await asyncio.wait_for( + process.wait(), + timeout=self._nunchi_feed_timeout_seconds, + ) + except TimeoutError: + with suppress(ProcessLookupError): + process.kill() + with suppress(Exception): + await asyncio.wait_for(process.wait(), timeout=5) + except Exception: + pass + + def kick_nunchi_feed(self, *, audience: str, scope: str) -> None: + """Fire the session-close nunchi ingest for one route (detached, body-free). + + The 10-minute cron dispatcher remains the owner of record; this kick only + removes the wait so the NEXT session's snapshot already includes the one + that just closed. Feed-side flock + seen-file keep cron overlap idempotent. + """ + + feed = self._validated_nunchi_feed() + if feed is None: + return + try: + environment = self._nunchi_feed_environment(audience=audience, scope=scope) + except _LocalIndexError: + return + + async def _run() -> None: + try: + spawned = await asyncio.create_subprocess_exec( + str(feed), + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + env=environment, + start_new_session=True, + ) + except (OSError, ValueError): + return + await self._drain_nunchi_feed(spawned) + + try: + task = asyncio.get_running_loop().create_task(_run()) + except RuntimeError: + return + self._nunchi_tasks.add(task) + task.add_done_callback(self._nunchi_tasks.discard) + def _index_environment( self, *, @@ -270,6 +405,7 @@ async def write_once(self, *, job_id: str) -> DistillJob: if audience is None or scope is None: raise ValueError("local sink job has no safe audience route") await self.refresh_route(audience=audience, scope=scope) + self.kick_nunchi_feed(audience=audience, scope=scope) except asyncio.CancelledError: await self._fail( claimed, diff --git a/bridge/tests/test_distill_local_worker.py b/bridge/tests/test_distill_local_worker.py index bdd35bdb..b0184eac 100644 --- a/bridge/tests/test_distill_local_worker.py +++ b/bridge/tests/test_distill_local_worker.py @@ -245,3 +245,133 @@ async def test_lifecycle_loop_drives_pending_local_work(tmp_path: Path) -> None: await asyncio.wait_for(task, timeout=2) assert journal.get(job.job_id).local_sink_status is DistillLocalSinkStatus.DONE + + +@pytest.mark.anyio +async def test_successful_sink_kicks_scoped_nunchi_feed(tmp_path: Path) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = await extracted_job(journal) + marker = tmp_path / "kick-env" + feed = tmp_path / "piri-feed.sh" + feed.write_text( + "#!/bin/sh\n" + "{\n" + "echo \"child=${CCC_NUNCHI_SCOPED_CHILD:-}\"\n" + "echo \"scope=${CCC_NUNCHI_AUDIENCE_SCOPE:-}\"\n" + "echo \"kind=${CCC_NUNCHI_AUDIENCE_KIND:-}\"\n" + "echo \"db=${NUNCHI_DB:-}\"\n" + "echo \"sessions=${PIR_SESSIONS_DIR:-}\"\n" + "echo \"token=${TELEGRAM_BOT_TOKEN:-}\"\n" + f"}} > {marker}\n", + ) + feed.chmod(0o700) + audience_root = tmp_path / "audiences" + worker = CodexDistillLocalSinkWorker( + journal, + audience_root=audience_root, + owner_token="local-worker", + nunchi_feed_path=feed, + environment={ + "HOME": str(tmp_path / "home"), + "PATH": "/usr/local/bin:/usr/bin:/bin", + "TELEGRAM_BOT_TOKEN": "RAW_TELEGRAM_TOKEN_MUST_NOT_CROSS", + }, + ) + + result = await worker.write_once(job_id=job.job_id) + + assert result.local_sink_status is DistillLocalSinkStatus.DONE + if worker._nunchi_tasks: + await asyncio.gather(*worker._nunchi_tasks) + env = dict( + line.split("=", 1) for line in marker.read_text().splitlines() + ) + assert env["child"] == "1" + assert env["scope"] == "private-0123456789abcdef0123456789abcdef" + assert env["kind"] == "private" + assert env["db"] == str( + audience_root + / "private-0123456789abcdef0123456789abcdef" + / "nunchi" + / "facts.db" + ) + assert env["sessions"] == str( + audience_root + / "private-0123456789abcdef0123456789abcdef" + / "piri" + / "sessions" + ) + assert env["token"] == "" + + +@pytest.mark.anyio +async def test_missing_nunchi_feed_skips_kick_and_sink_still_done( + tmp_path: Path, +) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = await extracted_job(journal) + worker = CodexDistillLocalSinkWorker( + journal, + audience_root=tmp_path / "audiences", + owner_token="local-worker", + nunchi_feed_path=tmp_path / "absent" / "piri-feed.sh", + ) + + result = await worker.write_once(job_id=job.job_id) + + assert result.local_sink_status is DistillLocalSinkStatus.DONE + if worker._nunchi_tasks: + await asyncio.gather(*worker._nunchi_tasks) + assert not (tmp_path / "absent").exists() + + +@pytest.mark.anyio +async def test_unsafe_nunchi_feed_is_never_executed(tmp_path: Path) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = await extracted_job(journal) + marker = tmp_path / "must-not-exist" + target = tmp_path / "unsafe-target.sh" + target.write_text(f"#!/bin/sh\ntouch {marker}\n") + target.chmod(0o700) + feed = tmp_path / "piri-feed.sh" + feed.symlink_to(target) + worker = CodexDistillLocalSinkWorker( + journal, + audience_root=tmp_path / "audiences", + owner_token="local-worker", + nunchi_feed_path=feed, + ) + + result = await worker.write_once(job_id=job.job_id) + + assert result.local_sink_status is DistillLocalSinkStatus.DONE + if worker._nunchi_tasks: + await asyncio.gather(*worker._nunchi_tasks) + assert not marker.exists() + + +@pytest.mark.anyio +async def test_failing_nunchi_feed_never_affects_the_sink_job( + tmp_path: Path, +) -> None: + journal = DistillJournal(tmp_path / "journal") + journal.initialize() + job = await extracted_job(journal) + feed = tmp_path / "piri-feed.sh" + feed.write_text("#!/bin/sh\nexit 9\n") + feed.chmod(0o700) + worker = CodexDistillLocalSinkWorker( + journal, + audience_root=tmp_path / "audiences", + owner_token="local-worker", + nunchi_feed_path=feed, + ) + + result = await worker.write_once(job_id=job.job_id) + + assert result.local_sink_status is DistillLocalSinkStatus.DONE + if worker._nunchi_tasks: + await asyncio.gather(*worker._nunchi_tasks)