| <cwd> | <YYYY-MM-DD> | source: pi] # so post-mine search results are self-identifying (pi vs opencode vs other). # # Dedup: mempalace convos mode keys on source_file (absolute staging path). # Staging paths are deterministic per pi session UUID, and the export copies # the source session's mtime onto the staged file, so re-runs are idempotent # until session content actually changes. A GROWN session is purged and # refiled for that source_file by the miner, so re-feeding a live session # refreshes its drawers instead of duplicating them. # # Staging location: source_file dedup keys on the staged path, so if the stage # is wiped the palace is left with drawers whose source files look deleted. # `mempalace sync` prunes exactly those — but only within the scope it is # given. Measured on this layout: scoped at the palace root the staged sources # are in scope (kept 651), while a wing-only sync reports them out_of_scope and # leaves them alone. So the data loss is conditional on how sync is invoked, # which is far too thin a margin to rely on. # # The stage therefore defaults NEXT TO THE PALACE (<palace-root>/pi-stage, # resolved the way mempalace itself resolves the palace: $MEMPALACE_PALACE_PATH # → $MEMPAL_PALACE_PATH → ~/.mempalace/config.json → ~/.mempalace/palace). # # That makes the invariant structural rather than documented: the stage and the # dedup keys that reference it share one lifetime, so the dangerous state — # palace survives, stage does not — can no longer be reached by wiping # something that merely looks disposable. A cache dir (the obvious choice, and # the old default) is exactly wrong here: it persists just long enough to look # correct, then takes the memories with it. Override with MEMPALACE_PI_STAGE # only if the target is at least as durable as the palace. # # In remote mode the local stage is only a shipping buffer — dedup lives on the # server, keyed by the server-side inbox path — so its durability is moot there. # # Session filter: two gates, both required. # 1. --min-messages <N> user+assistant turns (default 4). Tool loops inflate # assistant turns fast in pi, so a real working session clears this # easily; a single abandoned prompt does not. # 2. --min-assistant-chars <N> characters of assistant *text* (default 1000), # excluding tool results. Assistant volume, not total volume: pi expands # skills/context into the user prompt, so an abandoned session can carry a # 13k-char "user" message answered with "Ready. What would you like to # work on?" — total size says substantial, assistant size correctly says # nothing happened. # # Usage: # mempalace-pi-session # mempalace-pi-session --prepare # mempalace-pi-session --mode remote # mempalace-pi-session --wing <name> # mempalace-pi-session --session <uuid-prefix> # mempalace-pi-session --since 2026-04-01 # mempalace-pi-session --min-messages 6 # mempalace-pi-session --dry-run # mempalace-pi-session --help # # Exit codes: # 0 success (including "nothing qualified", "another run holds the lock", # and "palace held by a live session") # 1 usage / argument error # 2 pi sessions dir missing # 3 mempalace CLI not installed / rsync missing in remote mode # 4 mine failed # 5 remote transport failed (rsync or HTTP tools/call) # # Dependencies: bash, python3 (stdlib only), mempalace (v3.3.3+); # rsync + ssh in remote mode. set -euo pipefail # HOME can legitimately be unset: `docker run --entrypoint="" <image>` inherits # no HOME when the image config declares none (pi-devbox's does not — HOME is # normally set by its entrypoint, which --entrypoint="" skips), and every # default below is HOME-anchored under `set -u`, so the script died at line 1 of # real work with "HOME: unbound variable". Derive it from the passwd database — # exactly what python's expanduser() falls back to — so the script, and # especially the palace-free --self-test, runs in a bare container too. # pi-devbox v1.8.0 lost a release to this same "the image sets HOME" assumption. : "${HOME:=$(python3 -c 'import os, pwd; print(pwd.getpwuid(os.getuid()).pw_dir)' 2>/dev/null || echo /tmp)}" export HOME # ── Defaults ───────────────────────────────────────────────────────── AGENT="${USER:-mempalace}" WING="wing_conversations" SESSION_ID="" SINCE="" MIN_MESSAGES=4 MIN_ASSISTANT_CHARS=1000 DRY_RUN=0 DO_REPAIR=0 PREPARE_ONLY=0 SELF_TEST=0 MODE="auto" REASON="" PI_SESSIONS_DIR="${PI_SESSIONS_DIR:-$HOME/.pi/agent/sessions}" # Resolve the palace ROOT (the dir holding palace/, knowledge_graph.sqlite3, # config.json) using mempalace's own precedence, so the stage lands next to # whichever palace this host actually feeds. Mirrors config.py:palace_path() # (env → config.json → default) and takes the parent. Only evaluated when # MEMPALACE_PI_STAGE is unset, so the common path costs nothing. palace_root() { python3 - <<'PY' 2>/dev/null || echo "$HOME/.mempalace" import json, os p = os.environ.get("MEMPALACE_PALACE_PATH") or os.environ.get("MEMPAL_PALACE_PATH") if p: p = os.path.abspath(os.path.expanduser(p)) else: cfg = os.path.expanduser("~/.mempalace/config.json") p = None if os.path.exists(cfg): try: with open(cfg) as fh: v = json.load(fh).get("palace_path") p = os.path.expanduser(v) if v else None except Exception: p = None p = p or os.path.expanduser("~/.mempalace/palace") print(os.path.dirname(p.rstrip("/"))) PY } STAGE_ROOT="${MEMPALACE_PI_STAGE:-$(palace_root)/pi-stage}" # Remote transport (see TRANSPORTS in the header) REMOTE_URL="${MEMPALACE_REMOTE_URL:-}" REMOTE_TOKEN="${MEMPALACE_REMOTE_TOKEN:-}" SSH_TARGET="${MEMPALACE_PI_SSH_TARGET:-}" SSH_CONFIG="${MEMPALACE_PI_SSH_CONFIG:-}" REMOTE_PATH="${MEMPALACE_PI_REMOTE_PATH:-/data/feed}" DEVICE="${MEMPALACE_PI_DEVICE:-$(hostname)}" # ── Usage ──────────────────────────────────────────────────────────── usage() { cat <<'EOF' mempalace-pi-session — mine pi coding-agent session history into MemPalace Usage: mempalace-pi-session [options] Options: --wing <name> Target wing (default: wing_conversations) --session <prefix> Export one session only (match on UUID prefix) --since <YYYY-MM-DD> Only sessions last modified on/after this date --min-messages <N> Skip sessions with fewer than N user+assistant turns (default: 4) --min-assistant-chars <N> Skip sessions with fewer than N characters of assistant text, tool results excluded (default: 1000). Catches abandoned sessions whose bulk is injected skill/context text in the user prompt. --agent <name> Agent name recorded on drawers (default: $USER) --sessions-dir <path> Path to pi sessions dir (default: $PI_SESSIONS_DIR or ~/.pi/agent/sessions) --stage <path> Staging root (default: $MEMPALACE_PI_STAGE, else <palace-root>/pi-stage — next to the palace, so the stage cannot be wiped independently of the dedup keys that point at it). Exports go in <root>/<wing>. See "Staging location" in the header before moving it. --mode <m> auto|local|remote (default: auto — remote when $MEMPALACE_REMOTE_URL is set) --prepare Export + stage (+ rsync in remote mode), print MINE_SOURCE=<path>, and stop without opening the palace. For callers that will do the mine themselves through a live MCP connection. --reason <label> Label this run in its output (e.g. shutdown, tick, container-start). Useful when triggers log to a file. --dry-run Export + list; do not mine into palace. Each session is tagged [NEW] or [SKIP] based on whether its source_file is already in the palace. In remote mode the tag is [?]: dedup is decided by the palace host, which this machine's local palace copy cannot answer. --self-test Run the remote-mine response classifier against recorded MCP responses and exit. Needs no palace, no network and no sessions dir. --repair Run `mempalace repair` after mining (opt-in). WARNING: repair does a destructive in-place HNSW rebuild. If it races a live MCP connection or crashes mid-rebuild, it can wipe the collection. Only pass this from a quiet, interactive context. Not safe for unattended cron/launchd schedules. --no-repair (Deprecated; no-repair is now the default.) -h, --help Show this help Idempotency: Re-running on the same corpus is safe. The export step writes every qualifying session to the cache; the mine step dedups by source_file so already-filed sessions are skipped without re-embedding. Transcript shape per session: - Synthetic header as first user turn: [session: <title> | <cwd> | <YYYY-MM-DD> | source: pi] - User/assistant messages extracted from pi JSONL `message` entries - Assistant toolCall blocks → Claude Code `tool_use` blocks - `toolResult` role messages → `tool_result` blocks (folded back into the assistant turn by the normalizer) - `bashExecution`, `custom(display=true)`, `branchSummary`, `compactionSummary` → rendered as text annotations - `thinking` content blocks → dropped (noise) - Image content blocks → dropped (palace embeds text only) Dedup: - source_file = absolute staging path (deterministic per pi session UUID) - Re-runs skip unchanged sessions; a GROWN session (mtime changed) has its old drawers purged and is refiled, so re-feeding a live session refreshes rather than duplicates. - To force re-mining, delete the staging dir: rm -rf <palace-root>/pi-stage/<wing>/ That forces a refile — but do NOT run `mempalace sync` while the stage is missing, or the drawers mined from it get pruned instead. Rationale: Two complementary paths feed the palace from pi, and they cover different failure modes: - The pi mempalace bridge extension (extensions/pi/mempalace.ts) drives this script with --prepare on session_shutdown and on a debounced agent_settled, then mines through its own live MCP connection. That is the primary path: it needs no scheduling and it is the only way to write while a session holds the palace. - Running this script directly is the batch/recovery path: a container-start or host-level catch-up that picks up transcripts nothing mined at the time — notably after a SIGKILL, where no pi handler runs at all. It reads the durable on-disk JSONL, so it does not care whether the session that produced it exited cleanly. EOF } # ── Remote mine over MCP ───────────────────────────────────────────── # Usage: run_remote_mine <url> <token> <source> <wing> <agent> # run_remote_mine --self-test # # WHY THIS IS A FUNCTION WITH A SELF-TEST: MCP answers a hard tool failure with # HTTP 200 and a JSON-RPC *result* whose content[].text holds the tool's own # JSON as an ESCAPED STRING. This code used to decide success with # `'"error"' in body`, which can never match those bytes (they are \"error\"), # so on 2026-08-15 a mine that failed with # {"success": false, "error": "source directory not found: '/data/feed/...'"} # was reported as "Done. Wing updated." and nothing was filed. A silent # false success in a feeder is worse than a crash: the only artifact says it # worked. The fixtures below pin that exact body so it cannot come back. run_remote_mine() { python3 - "$@" <<'PY' import json, sys, urllib.error, urllib.request def classify(body): """Return (ok, note) for an MCP tools/call response body. ok=False means the mine demonstrably failed. note carries the reason, or — when ok is True — an "unverified" caveat if the response contained no JSON tool payload to adjudicate. Never claim more than the bytes support. """ try: env = json.loads(body) except ValueError: return False, "response was not JSON: " + body[:200].replace("\n", " ") if not isinstance(env, dict): return False, "response was not a JSON object" if env.get("error") is not None: # JSON-RPC transport-level error return False, "JSON-RPC error: " + json.dumps(env["error"])[:300] result = env.get("result") if not isinstance(result, dict): return False, "response carried no result object" if result.get("isError"): return False, "MCP isError set: " + json.dumps(result.get("content"))[:300] saw_payload = False for item in result.get("content") or []: text = item.get("text") if isinstance(item, dict) else None if not isinstance(text, str): continue try: payload = json.loads(text) # the escaped inner JSON except ValueError: continue # plain prose content: nothing to judge if not isinstance(payload, dict): continue saw_payload = True if payload.get("success") is False: return False, str(payload.get("error") or "tool reported success=false") if payload.get("error"): return False, str(payload["error"]) if not saw_payload: return True, "unverified: no JSON tool payload in the response" return True, "" FIXTURES = [ # 1. The real 2026-08-15 failure: HTTP 200, JSON-RPC result, tool failed. ('{"jsonrpc": "2.0", "id": 1, "result": {"content": [{"type": "text", ' '"text": "{\\n \\"success\\": false,\\n \\"error\\": \\"source directory ' 'not found: \'/data/feed/emb-7kj4vr4g\'\\"\\n}"}]}}', False), # 2. A real success. ('{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":' '"{\\"success\\": true, \\"mode\\": \\"convos\\", \\"output\\": \\"Drawers filed: 12\\"}"}]}}', True), # 3. JSON-RPC level error (bad method, auth rejected at protocol level). ('{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"Method not found"}}', False), # 4. MCP tool-level isError flag. ('{"jsonrpc":"2.0","id":1,"result":{"isError":true,"content":[{"type":"text","text":"boom"}]}}', False), # 5. Not JSON at all (proxy error page, 502 HTML). ('<html><body>502 Bad Gateway</body></html>', False), # 6. Success-shaped envelope with prose content: cannot be called a failure. ('{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"mined 3 files"}]}}', True), ] args = sys.argv[1:] if args[:1] == ["--self-test"]: failures = 0 for i, (body, want_ok) in enumerate(FIXTURES, 1): got_ok, note = classify(body) if got_ok != want_ok: failures += 1 print(f" [{'ok ' if got_ok == want_ok else 'FAIL'}] fixture {i}: " f"want_ok={want_ok} got_ok={got_ok} note={note[:70]!r}") # Regression guard: the detector this replaced must be shown blind to #1. old_detector_sees_it = '"error"' in FIXTURES[0][0] if old_detector_sees_it: failures += 1 print(f" [{'ok ' if not old_detector_sees_it else 'FAIL'}] regression guard: " f"substring detector sees fixture 1? {old_detector_sees_it} (must be False)") print("SELF-TEST " + ("FAILED" if failures else "PASSED")) sys.exit(1 if failures else 0) url, token, source, wing, agent = args[:5] payload = json.dumps({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "mempalace_mine", "arguments": {"source": source, "mode": "convos", "wing": wing, "agent": agent}, }, }).encode() headers = {"Content-Type": "application/json", "Accept": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" try: with urllib.request.urlopen( urllib.request.Request(url, data=payload, headers=headers), timeout=900 ) as resp: body = resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace")[:500] if hasattr(exc, "read") else "" print(f"error: remote mine transport failed: HTTP {exc.code} {exc.reason} {detail}", file=sys.stderr) sys.exit(1) except urllib.error.URLError as exc: print(f"error: remote mine transport failed: {exc}", file=sys.stderr) sys.exit(1) print(body[:4000]) ok, note = classify(body) if not ok: print(f"error: remote mine reported failure: {note}", file=sys.stderr) sys.exit(1) if note: print(f"warning: remote mine {note}", file=sys.stderr) sys.exit(0) PY } # ── Parse args ─────────────────────────────────────────────────────── while [[ $# -gt 0 ]]; do case "$1" in -h|--help) usage; exit 0 ;; --wing) WING="${2:-}"; shift 2 ;; --session) SESSION_ID="${2:-}"; shift 2 ;; --since) SINCE="${2:-}"; shift 2 ;; --min-messages) MIN_MESSAGES="${2:-}"; shift 2 ;; --min-assistant-chars) MIN_ASSISTANT_CHARS="${2:-}"; shift 2 ;; --stage) STAGE_ROOT="${2:-}"; shift 2 ;; --mode) MODE="${2:-}"; shift 2 ;; --prepare) PREPARE_ONLY=1; shift ;; --reason) REASON="${2:-}"; shift 2 ;; --agent) AGENT="${2:-}"; shift 2 ;; --sessions-dir) PI_SESSIONS_DIR="${2:-}"; shift 2 ;; --dry-run) DRY_RUN=1; shift ;; --self-test) SELF_TEST=1; shift ;; --repair) DO_REPAIR=1; shift ;; --no-repair) shift ;; # deprecated alias; no-repair is the default --) shift; break ;; -*) echo "error: unknown option: $1" >&2; usage >&2; exit 1 ;; *) echo "error: unexpected arg: $1" >&2; exit 1 ;; esac done # The classifier self-test is pure logic: no palace, no network, no sessions # dir. Dispatch before preflight so it runs anywhere, including in CI. if [[ $SELF_TEST -eq 1 ]]; then echo "mempalace-pi-session --self-test: remote-mine response classifier" run_remote_mine --self-test exit $? fi # ── Preflight ──────────────────────────────────────────────────────── if [[ ! -d "$PI_SESSIONS_DIR" ]]; then echo "error: pi sessions dir not found at $PI_SESSIONS_DIR" >&2 echo " override with --sessions-dir <path> or PI_SESSIONS_DIR env var" >&2 exit 2 fi case "$MODE" in auto) if [[ -n "$REMOTE_URL" ]]; then MODE="remote"; else MODE="local"; fi ;; local|remote) ;; *) echo "error: --mode must be auto|local|remote" >&2; exit 1 ;; esac # The mempalace CLI is only needed when WE do the mine. --prepare never opens # the palace, and remote mode talks to the server over HTTP. if [[ $PREPARE_ONLY -eq 0 && "$MODE" == "local" ]] && ! command -v mempalace >/dev/null 2>&1; then echo "error: mempalace CLI not found in PATH" >&2 exit 3 fi if [[ "$MODE" == "remote" ]]; then command -v rsync >/dev/null 2>&1 || { echo "error: rsync not found (needed for --mode remote)" >&2; exit 3; } if [[ -z "$SSH_TARGET" ]]; then echo "error: MEMPALACE_PI_SSH_TARGET unset (needed for --mode remote)" >&2 exit 1 fi # The devbox generates a dedicated LAN-jump key/config; prefer it if present. if [[ -z "$SSH_CONFIG" && -f "$HOME/.ssh-local/config" ]]; then SSH_CONFIG="$HOME/.ssh-local/config" fi # Remote mode names the same inbox twice: where rsync PUTS the files, and # what the SERVER is told to mine. They may legitimately differ # (containerized server: host dir bind-mounted elsewhere), but when they # differ by accident rsync still succeeds and only the mine fails — the # 2026-08-15 /data/feed incident, where transcripts shipped for hours and # were filed nowhere. Warn in PREFLIGHT so --dry-run and --prepare see it # too, not just a full run that gets as far as shipping. SHIP_PATH="$SSH_TARGET" [[ "$SHIP_PATH" == *:* ]] && SHIP_PATH="${SHIP_PATH##*:}" if [[ "${SHIP_PATH%/}" != "${REMOTE_PATH%/}" ]]; then echo "note: shipping to '${SHIP_PATH%/}/$DEVICE' but asking the server to mine" echo " '${REMOTE_PATH%/}/$DEVICE'. Correct only if the palace server sees" echo " '${SHIP_PATH%/}' at '${REMOTE_PATH%/}' (containerized server with a bind" echo " mount). A NATIVE server (systemd unit / uv tool) sees host paths —" echo " then set MEMPALACE_PI_REMOTE_PATH='${SHIP_PATH%/}'." fi fi for _n in MIN_MESSAGES MIN_ASSISTANT_CHARS; do if ! [[ "${!_n}" =~ ^[0-9]+$ ]]; then _flag="--$(printf '%s' "${_n,,}" | tr '_' '-')" echo "error: $_flag must be an integer" >&2 exit 1 fi done # ── Staging dir ────────────────────────────────────────────────────── # Deterministic per-wing path so source_file dedup works across re-runs. See # "Staging location" in the header for why this should not be disposable. STAGE="${STAGE_ROOT%/}/$WING" mkdir -p "$STAGE" [[ -n "$REASON" ]] && echo "mempalace-pi-session [$REASON] mode=$MODE stage=$STAGE" # ── Single-writer guard ────────────────────────────────────────────── # Non-blocking: overlapping triggers (a session_shutdown landing on top of a # debounced mid-session run) must not queue or race. Losing a run is harmless # — the next one re-exports from scratch. exec 9>"${STAGE_ROOT%/}/.lock" if command -v flock >/dev/null 2>&1 && ! flock -n 9; then echo "another mempalace-pi-session run holds the lock; skipping" exit 0 fi # ── Export sessions (Python heredoc) ──────────────────────────────── # Parses pi JSONL files and writes Claude Code JSONL per session into $STAGE. # Also classifies each export as NEW/ALREADY FILED (by source_file lookup) # so --dry-run reports the real mine-set size. Classification is advisory; # `mempalace mine --mode convos` is still the authoritative dedup. export_count=$(python3 - "$PI_SESSIONS_DIR" "$STAGE" "$SESSION_ID" "$SINCE" "$MIN_MESSAGES" "$MIN_ASSISTANT_CHARS" "$MODE" <<'PY' import json, os, sqlite3, sys from datetime import datetime, timezone from pathlib import Path sessions_dir, stage, session_filter, since, min_messages, min_assistant_chars, mode = sys.argv[1:8] min_messages = int(min_messages) min_assistant_chars = int(min_assistant_chars) stage = Path(stage) sessions_dir = Path(sessions_dir) # Convert --since YYYY-MM-DD to epoch seconds (comparing against file mtime) since_epoch = None if since: try: since_epoch = datetime.strptime(since, "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp() except ValueError: print(f"error: --since must be YYYY-MM-DD, got {since!r}", file=sys.stderr) sys.exit(1) # ── Load palace's already-filed source_files (best-effort, read-only) ── # already_filed is None in remote mode: unknown, not empty. The dedup that # matters happens on the PALACE HOST and is keyed on the REMOTE inbox path, so # this machine's local palace file cannot answer the question — and answering # it anyway is how a preview comes to say "6 already filed" about a palace it # is not feeding (2026-08-15). An honest "[?]" beats a confident wrong number. already_filed = None if mode == "remote" else set() # Mirror mempalace's own resolution order (config.py): MEMPALACE_PALACE_PATH, # then the legacy MEMPAL_PALACE_PATH, then the default. NOT "MEMPALACE_PATH" — # that name is not a mempalace concept, and reading it silently degraded this # NEW/SKIP preview to "everything is new" wherever some other tool had set it. palace_path = ( os.environ.get("MEMPALACE_PALACE_PATH") or os.environ.get("MEMPAL_PALACE_PATH") or os.path.expanduser("~/.mempalace/palace") ) chroma_db = Path(palace_path) / "chroma.sqlite3" if already_filed is not None and chroma_db.is_file(): try: pcon = sqlite3.connect(f"file:{chroma_db}?mode=ro", uri=True) for (sf,) in pcon.execute( "SELECT DISTINCT string_value FROM embedding_metadata " "WHERE key='source_file' AND string_value LIKE ?", (f"{stage}%",), ): if sf: already_filed.add(sf) pcon.close() except sqlite3.Error: pass # palace unreachable → miner will dedup def extract_text(content): """Flatten a message content (string | list-of-blocks) to plain text. Drops image + thinking blocks; keeps text + renders toolCall/toolResult stubs inline. Returns ("", [tool_uses], [tool_results]) where tool_uses are collected for assistant messages and tool_results for toolResult messages. """ if isinstance(content, str): return content, [], [] if not isinstance(content, list): return "", [], [] text_parts = [] tool_uses = [] for block in content: if not isinstance(block, dict): continue bt = block.get("type") if bt == "text": t = block.get("text", "") if t: text_parts.append(t) elif bt == "thinking": # Drop reasoning content — high-noise, low-signal for search. continue elif bt == "image": # Palace is text-only. continue elif bt == "toolCall": tool_uses.append({ "type": "tool_use", "id": block.get("id") or "", "name": block.get("name") or "tool", "input": block.get("arguments") or {}, }) return "\n".join(text_parts), tool_uses, [] def load_session(path: Path): """Parse a pi JSONL session file. Returns (header, entries) or None.""" try: with path.open("r", encoding="utf-8") as f: lines = [ln for ln in f.read().splitlines() if ln.strip()] except OSError: return None if not lines: return None try: header = json.loads(lines[0]) except json.JSONDecodeError: return None if header.get("type") != "session": return None entries = [] for ln in lines[1:]: try: entries.append(json.loads(ln)) except json.JSONDecodeError: continue return header, entries def derive_title(entries, fallback: str) -> str: """Prefer session_info.name; else truncated first user message.""" # session_info entries: most-recent wins name = None for e in entries: if e.get("type") == "session_info" and e.get("name"): name = e["name"] if name: return name[:120] for e in entries: if e.get("type") != "message": continue msg = e.get("message") or {} if msg.get("role") != "user": continue text, _, _ = extract_text(msg.get("content")) text = " ".join(text.split()) # collapse whitespace if text: return (text[:80] + "…") if len(text) > 80 else text return fallback # Discover session files paths = sorted(sessions_dir.rglob("*.jsonl")) if session_filter: paths = [p for p in paths if session_filter in p.name] exported = 0 skipped_short = 0 skipped_quiet = 0 skipped_malformed = 0 skipped_already_filed = 0 for path in paths: try: mtime = path.stat().st_mtime except OSError: continue if since_epoch is not None and mtime < since_epoch: continue parsed = load_session(path) if parsed is None: skipped_malformed += 1 continue header, entries = parsed session_uuid = header.get("id") or path.stem cwd = header.get("cwd") or "?" header_ts = header.get("timestamp") or "" try: date_str = header_ts[:10] if header_ts else datetime.fromtimestamp( mtime, tz=timezone.utc).strftime("%Y-%m-%d") except Exception: date_str = datetime.fromtimestamp(mtime, tz=timezone.utc).strftime("%Y-%m-%d") # Count user+assistant message entries for the min-messages filter turn_count = sum( 1 for e in entries if e.get("type") == "message" and (e.get("message") or {}).get("role") in ("user", "assistant") ) if turn_count < min_messages: skipped_short += 1 continue title = derive_title(entries, fallback=session_uuid[:8]) assistant_chars = 0 out_lines = [] out_lines.append({ "type": "user", "message": { "content": f"[session: {title} | {cwd} | {date_str} | source: pi]" }, }) for e in entries: t = e.get("type") if t == "message": msg = e.get("message") or {} role = msg.get("role") if role == "user": text, _, _ = extract_text(msg.get("content")) if text.strip(): out_lines.append({"type": "user", "message": {"content": text}}) elif role == "assistant": text, tool_uses, _ = extract_text(msg.get("content")) assistant_chars += len(text.strip()) blocks = [] if text.strip(): blocks.append({"type": "text", "text": text}) blocks.extend(tool_uses) if not blocks: continue # Simplify single-text to string (matches mempalace-session). if len(blocks) == 1 and blocks[0].get("type") == "text": content = blocks[0]["text"] else: content = blocks out_lines.append({"type": "assistant", "message": {"content": content}}) elif role == "toolResult": text, _, _ = extract_text(msg.get("content")) tool_id = msg.get("toolCallId") or "" if not tool_id: continue out_lines.append({ "type": "human", "message": { "content": [{ "type": "tool_result", "tool_use_id": tool_id, "content": text or "(no output)", }], }, }) elif role == "bashExecution": # Rendered as a synthetic assistant annotation so the # command + output stay associated with the surrounding turn. cmd = msg.get("command") or "" out = msg.get("output") or "" exit_code = msg.get("exitCode") note = f"[user-bash] $ {cmd}\nexit={exit_code}\n{out}".strip() if note: out_lines.append({"type": "user", "message": {"content": note}}) elif role == "custom": if not msg.get("display"): continue text, _, _ = extract_text(msg.get("content")) if text.strip(): ctype = msg.get("customType") or "custom" out_lines.append({ "type": "user", "message": {"content": f"[custom:{ctype}] {text}"}, }) elif role in ("branchSummary", "compactionSummary"): summary = msg.get("summary") or "" if summary.strip(): out_lines.append({ "type": "user", "message": {"content": f"[{role}] {summary}"}, }) # thinking-only / empty messages silently dropped elif t in ( "model_change", "thinking_level_change", "compaction", "branch_summary", "label", "session_info", "custom", "custom_message", ): # Non-conversational entries: drop. (custom_message with # display=true could be included but we already get it via the # "custom" message role above when pi materializes one.) continue # Need at least 2 turns (header + one real turn) for the normalizer. if len(out_lines) < 2: skipped_short += 1 continue # Assistant *text* volume, tool results excluded: the signal that the # session actually did something, independent of how much injected # skill/context text inflated the user side. if assistant_chars < min_assistant_chars: skipped_quiet += 1 print( f" [QUIET] {path.name} ({turn_count} turns, {assistant_chars} assistant chars)", file=sys.stderr, ) continue out_path = stage / f"pi_{session_uuid}.jsonl" with out_path.open("w", encoding="utf-8") as f: for obj in out_lines: f.write(json.dumps(obj, ensure_ascii=False) + "\n") # Preserve session mtime on the staging file for dedup stability. try: os.utime(out_path, (mtime, mtime)) except OSError: pass exported += 1 if already_filed is None: is_filed = False # unknowable here; the palace host decides status = "? " else: is_filed = str(out_path) in already_filed if is_filed: skipped_already_filed += 1 status = "SKIP" if is_filed else "NEW " print(f" [{status}] {out_path.name} ({turn_count} turns)", file=sys.stderr) print(f"EXPORTED {exported}") print(f"ALREADY_FILED {-1 if already_filed is None else skipped_already_filed}") if skipped_short: print(f"SKIPPED_SHORT {skipped_short}", file=sys.stderr) if skipped_quiet: print(f"SKIPPED_QUIET {skipped_quiet}", file=sys.stderr) if skipped_malformed: print(f"SKIPPED_MALFORMED {skipped_malformed}", file=sys.stderr) PY ) # Parse counts from stdout count="$(printf '%s\n' "$export_count" | awk '/^EXPORTED / { print $2 }')" count="${count:-0}" already_filed="$(printf '%s\n' "$export_count" | awk '/^ALREADY_FILED / { print $2 }')" already_filed="${already_filed:-0}" # -1 means "unknown" (remote mode), so guard the arithmetic. if [[ "$already_filed" -lt 0 ]]; then to_file="$count"; else to_file=$(( count - already_filed )); fi if [[ "$count" -eq 0 ]]; then echo "no sessions qualified for export" exit 0 fi echo "" echo "Exported $count session(s) to $STAGE" if [[ "$already_filed" -lt 0 ]]; then # Remote mode: dedup lives on the palace host, keyed on the remote inbox # path. Do not translate "unknown" into a number. to_file="$count" echo " all $count shipped → the palace host dedups by source_file (remote mode:" echo " this machine cannot preview what it already holds)" else echo " $to_file new → will be filed on mine" echo " $already_filed already filed → will be skipped (dedup by source_file)" fi if [[ $DRY_RUN -eq 1 ]]; then echo "" if [[ "$already_filed" -lt 0 ]]; then echo "--dry-run: skipping ship+mine. A real run would ship $count session(s) to" echo " ${SSH_TARGET%/}/$DEVICE/ and let the palace host dedup them." elif [[ "$to_file" -eq 0 ]]; then echo "--dry-run: no new sessions to mine. A real run would skip all $count." else echo "--dry-run: skipping mine step. A real run would file $to_file new session(s)." fi exit 0 fi # ── Ship to the palace host (remote mode only) ─────────────────────── # mempalace_mine expands its source path in the SERVER process, so in remote # mode the exports have to physically exist over there. rsync --update is the # idempotent half; the mine is the other half. MINE_SOURCE="$STAGE" if [[ "$MODE" == "remote" ]]; then ssh_cmd="ssh" [[ -n "$SSH_CONFIG" ]] && ssh_cmd="ssh -F $SSH_CONFIG" echo "" echo "Shipping stage to ${SSH_TARGET%/}/$DEVICE/ ..." if ! rsync -a --update --no-owner --no-group \ -e "$ssh_cmd" \ --include='*.jsonl' --exclude='*' \ "$STAGE/" "${SSH_TARGET%/}/$DEVICE/"; then echo "error: rsync to ${SSH_TARGET%/}/$DEVICE/ failed" >&2 exit 5 fi MINE_SOURCE="${REMOTE_PATH%/}/$DEVICE" fi # ── Phase boundary ─────────────────────────────────────────────────── # --prepare hands the source path to the caller (the pi mempalace extension), # which mines it through the MCP client that already holds the palace. if [[ $PREPARE_ONLY -eq 1 ]]; then echo "" printf 'MINE_SOURCE=%s\n' "$MINE_SOURCE" exit 0 fi # ── Run the mine ───────────────────────────────────────────────────── echo "" echo "Mining into wing '$WING'..." if [[ "$MODE" == "remote" ]]; then if ! run_remote_mine "$REMOTE_URL" "$REMOTE_TOKEN" "$MINE_SOURCE" "$WING" "$AGENT"; then echo "error: remote mine failed" >&2 exit 5 fi else # Capture output so palace-level contention can be told apart from a real # failure. A live pi session holds the palace through its own mempalace-mcp, # and that session's extension mines what we just staged — so contention # means "already handled", not "broken". set +e mine_out="$(mempalace mine "$MINE_SOURCE" --mode convos --wing "$WING" --agent "$AGENT" 2>&1)" mine_rc=$? set -e printf '%s\n' "$mine_out" if [[ $mine_rc -ne 0 ]]; then if printf '%s' "$mine_out" | grep -q "is held by"; then echo "" echo "palace is held by a live session; it will mine the staged exports itself" exit 0 fi echo "error: mempalace mine failed" >&2 exit 4 fi fi # ── Repair index ───────────────────────────────────────────────────── if [[ $DO_REPAIR -eq 1 ]]; then echo "" echo "WARNING: --repair runs an in-place HNSW rebuild that has wiped" echo " live palaces on past runs. Proceeding in 3 seconds..." sleep 3 echo "Rebuilding HNSW index..." mempalace repair --yes fi echo "" echo "Done. Wing '$WING' updated. Remember to reconnect any live MCP sessions."

#!/usr/bin/env bash # mempalace-pi-session — mine pi coding-agent session history into MemPalace # # Pi persists every session (verbatim user/assistant turns + tool calls + tool # results) as newline-delimited JSONL under ~/.pi/agent/sessions/. Pi has no # upstream MemPalace integration and mempalace-toolkit's existing wrapper # (`mempalace-session`) only handles opencode's SQLite DB, so pi sessions are # currently invisible to the palace. # # Strategy (mirrors mempalace-session): # 1. Walk ~/.pi/agent/sessions/**/*.jsonl and export each qualifying session # to a Claude Code JSONL file (format the mempalace normalizer speaks). # 2. Stage exports under $MEMPALACE_PI_STAGE/ (default # /pi-stage/ — alongside the palace it feeds). # 3. Run `mempalace mine --mode convos` against the staging dir. # # TWO PHASES (--prepare), because the palace is single-writer # mempalace refuses a CLI mine while another process holds the palace: # "palace ... is held by PID (mempalace-mcp); wait for it to finish" # A live pi session ALWAYS has a holder — the mempalace extension's own # mempalace-mcp. So an unattended CLI mine only works when no session is # live (e.g. a container-start catch-up); during a session the mine must be # performed by the process that already holds the palace. Hence: # # --prepare export + stage (+ rsync in remote mode) and print # MINE_SOURCE=, without ever opening the palace. # (default) the above, then mine it ourselves. Contention is treated as # success-with-nothing-to-do, not failure: the holder's own # extension will mine what we staged. # # The pi mempalace extension drives exactly this: it runs --prepare on # session_shutdown and on a debounced agent_settled, then calls # mempalace_mine on MINE_SOURCE through its existing MCP client. # # TRANSPORTS (--mode, default auto) # local Mine into the local palace with the mempalace CLI. # remote $MEMPALACE_REMOTE_URL is set, so the palace lives on another host. # There is no remote-palace CLI — only the HTTP MCP server — and # mempalace_mine expands its source path in the SERVER process, so # the server cannot see this machine's staged exports. We therefore # rsync the stage into a per-device inbox on the palace host and ask # the server to mine its own local path. Requires # MEMPALACE_PI_SSH_TARGET (where to rsync) and # MEMPALACE_PI_REMOTE_PATH (what that inbox is called server-side). # # MEMPALACE_PI_REMOTE_PATH is the path AS THE SERVER PROCESS SEES # IT, and the default (/data/feed) assumes a CONTAINERIZED server # with the inbox bind-mounted there. A NATIVE server (systemd unit / # uv tool / plain `mempalace serve`) sees host paths, so there it # must equal the path half of MEMPALACE_PI_SSH_TARGET. Get this # wrong and rsync still succeeds while the mine fails with # "source directory not found" — so a mismatch between the two is # warned about at ship time, and the mine's own failure is now # detected properly (see classify() in run_remote_mine). # # Labelling: every exported transcript begins with a synthetic header # [session: | <cwd> | <YYYY-MM-DD> | source: pi] # so post-mine search results are self-identifying (pi vs opencode vs other). # # Dedup: mempalace convos mode keys on source_file (absolute staging path). # Staging paths are deterministic per pi session UUID, and the export copies # the source session's mtime onto the staged file, so re-runs are idempotent # until session content actually changes. A GROWN session is purged and # refiled for that source_file by the miner, so re-feeding a live session # refreshes its drawers instead of duplicating them. # # Staging location: source_file dedup keys on the staged path, so if the stage # is wiped the palace is left with drawers whose source files look deleted. # `mempalace sync` prunes exactly those — but only within the scope it is # given. Measured on this layout: scoped at the palace root the staged sources # are in scope (kept 651), while a wing-only sync reports them out_of_scope and # leaves them alone. So the data loss is conditional on how sync is invoked, # which is far too thin a margin to rely on. # # The stage therefore defaults NEXT TO THE PALACE (<palace-root>/pi-stage, # resolved the way mempalace itself resolves the palace: $MEMPALACE_PALACE_PATH # → $MEMPAL_PALACE_PATH → ~/.mempalace/config.json → ~/.mempalace/palace). # # That makes the invariant structural rather than documented: the stage and the # dedup keys that reference it share one lifetime, so the dangerous state — # palace survives, stage does not — can no longer be reached by wiping # something that merely looks disposable. A cache dir (the obvious choice, and # the old default) is exactly wrong here: it persists just long enough to look # correct, then takes the memories with it. Override with MEMPALACE_PI_STAGE # only if the target is at least as durable as the palace. # # In remote mode the local stage is only a shipping buffer — dedup lives on the # server, keyed by the server-side inbox path — so its durability is moot there. # # Session filter: two gates, both required. # 1. --min-messages <N> user+assistant turns (default 4). Tool loops inflate # assistant turns fast in pi, so a real working session clears this # easily; a single abandoned prompt does not. # 2. --min-assistant-chars <N> characters of assistant *text* (default 1000), # excluding tool results. Assistant volume, not total volume: pi expands # skills/context into the user prompt, so an abandoned session can carry a # 13k-char "user" message answered with "Ready. What would you like to # work on?" — total size says substantial, assistant size correctly says # nothing happened. # # Usage: # mempalace-pi-session # mempalace-pi-session --prepare # mempalace-pi-session --mode remote # mempalace-pi-session --wing <name> # mempalace-pi-session --session <uuid-prefix> # mempalace-pi-session --since 2026-04-01 # mempalace-pi-session --min-messages 6 # mempalace-pi-session --dry-run # mempalace-pi-session --help # # Exit codes: # 0 success (including "nothing qualified", "another run holds the lock", # and "palace held by a live session") # 1 usage / argument error # 2 pi sessions dir missing # 3 mempalace CLI not installed / rsync missing in remote mode # 4 mine failed # 5 remote transport failed (rsync or HTTP tools/call) # # Dependencies: bash, python3 (stdlib only), mempalace (v3.3.3+); # rsync + ssh in remote mode. set -euo pipefail # HOME can legitimately be unset: `docker run --entrypoint="" <image>` inherits # no HOME when the image config declares none (pi-devbox's does not — HOME is # normally set by its entrypoint, which --entrypoint="" skips), and every # default below is HOME-anchored under `set -u`, so the script died at line 1 of # real work with "HOME: unbound variable". Derive it from the passwd database — # exactly what python's expanduser() falls back to — so the script, and # especially the palace-free --self-test, runs in a bare container too. # pi-devbox v1.8.0 lost a release to this same "the image sets HOME" assumption. : "${HOME:=$(python3 -c 'import os, pwd; print(pwd.getpwuid(os.getuid()).pw_dir)' 2>/dev/null || echo /tmp)}" export HOME # ── Defaults ───────────────────────────────────────────────────────── AGENT="${USER:-mempalace}" WING="wing_conversations" SESSION_ID="" SINCE="" MIN_MESSAGES=4 MIN_ASSISTANT_CHARS=1000 DRY_RUN=0 DO_REPAIR=0 PREPARE_ONLY=0 SELF_TEST=0 MODE="auto" REASON="" PI_SESSIONS_DIR="${PI_SESSIONS_DIR:-$HOME/.pi/agent/sessions}" # Resolve the palace ROOT (the dir holding palace/, knowledge_graph.sqlite3, # config.json) using mempalace's own precedence, so the stage lands next to # whichever palace this host actually feeds. Mirrors config.py:palace_path() # (env → config.json → default) and takes the parent. Only evaluated when # MEMPALACE_PI_STAGE is unset, so the common path costs nothing. palace_root() { python3 - <<'PY' 2>/dev/null || echo "$HOME/.mempalace" import json, os p = os.environ.get("MEMPALACE_PALACE_PATH") or os.environ.get("MEMPAL_PALACE_PATH") if p: p = os.path.abspath(os.path.expanduser(p)) else: cfg = os.path.expanduser("~/.mempalace/config.json") p = None if os.path.exists(cfg): try: with open(cfg) as fh: v = json.load(fh).get("palace_path") p = os.path.expanduser(v) if v else None except Exception: p = None p = p or os.path.expanduser("~/.mempalace/palace") print(os.path.dirname(p.rstrip("/"))) PY } STAGE_ROOT="${MEMPALACE_PI_STAGE:-$(palace_root)/pi-stage}" # Remote transport (see TRANSPORTS in the header) REMOTE_URL="${MEMPALACE_REMOTE_URL:-}" REMOTE_TOKEN="${MEMPALACE_REMOTE_TOKEN:-}" SSH_TARGET="${MEMPALACE_PI_SSH_TARGET:-}" SSH_CONFIG="${MEMPALACE_PI_SSH_CONFIG:-}" REMOTE_PATH="${MEMPALACE_PI_REMOTE_PATH:-/data/feed}" DEVICE="${MEMPALACE_PI_DEVICE:-$(hostname)}" # ── Usage ──────────────────────────────────────────────────────────── usage() { cat <<'EOF' mempalace-pi-session — mine pi coding-agent session history into MemPalace Usage: mempalace-pi-session [options] Options: --wing <name> Target wing (default: wing_conversations) --session <prefix> Export one session only (match on UUID prefix) --since <YYYY-MM-DD> Only sessions last modified on/after this date --min-messages <N> Skip sessions with fewer than N user+assistant turns (default: 4) --min-assistant-chars <N> Skip sessions with fewer than N characters of assistant text, tool results excluded (default: 1000). Catches abandoned sessions whose bulk is injected skill/context text in the user prompt. --agent <name> Agent name recorded on drawers (default: $USER) --sessions-dir <path> Path to pi sessions dir (default: $PI_SESSIONS_DIR or ~/.pi/agent/sessions) --stage <path> Staging root (default: $MEMPALACE_PI_STAGE, else <palace-root>/pi-stage — next to the palace, so the stage cannot be wiped independently of the dedup keys that point at it). Exports go in <root>/<wing>. See "Staging location" in the header before moving it. --mode <m> auto|local|remote (default: auto — remote when $MEMPALACE_REMOTE_URL is set) --prepare Export + stage (+ rsync in remote mode), print MINE_SOURCE=<path>, and stop without opening the palace. For callers that will do the mine themselves through a live MCP connection. --reason <label> Label this run in its output (e.g. shutdown, tick, container-start). Useful when triggers log to a file. --dry-run Export + list; do not mine into palace. Each session is tagged [NEW] or [SKIP] based on whether its source_file is already in the palace. In remote mode the tag is [?]: dedup is decided by the palace host, which this machine's local palace copy cannot answer. --self-test Run the remote-mine response classifier against recorded MCP responses and exit. Needs no palace, no network and no sessions dir. --repair Run `mempalace repair` after mining (opt-in). WARNING: repair does a destructive in-place HNSW rebuild. If it races a live MCP connection or crashes mid-rebuild, it can wipe the collection. Only pass this from a quiet, interactive context. Not safe for unattended cron/launchd schedules. --no-repair (Deprecated; no-repair is now the default.) -h, --help Show this help Idempotency: Re-running on the same corpus is safe. The export step writes every qualifying session to the cache; the mine step dedups by source_file so already-filed sessions are skipped without re-embedding. Transcript shape per session: - Synthetic header as first user turn: [session: <title> | <cwd> | <YYYY-MM-DD> | source: pi] - User/assistant messages extracted from pi JSONL `message` entries - Assistant toolCall blocks → Claude Code `tool_use` blocks - `toolResult` role messages → `tool_result` blocks (folded back into the assistant turn by the normalizer) - `bashExecution`, `custom(display=true)`, `branchSummary`, `compactionSummary` → rendered as text annotations - `thinking` content blocks → dropped (noise) - Image content blocks → dropped (palace embeds text only) Dedup: - source_file = absolute staging path (deterministic per pi session UUID) - Re-runs skip unchanged sessions; a GROWN session (mtime changed) has its old drawers purged and is refiled, so re-feeding a live session refreshes rather than duplicates. - To force re-mining, delete the staging dir: rm -rf <palace-root>/pi-stage/<wing>/ That forces a refile — but do NOT run `mempalace sync` while the stage is missing, or the drawers mined from it get pruned instead. Rationale: Two complementary paths feed the palace from pi, and they cover different failure modes: - The pi mempalace bridge extension (extensions/pi/mempalace.ts) drives this script with --prepare on session_shutdown and on a debounced agent_settled, then mines through its own live MCP connection. That is the primary path: it needs no scheduling and it is the only way to write while a session holds the palace. - Running this script directly is the batch/recovery path: a container-start or host-level catch-up that picks up transcripts nothing mined at the time — notably after a SIGKILL, where no pi handler runs at all. It reads the durable on-disk JSONL, so it does not care whether the session that produced it exited cleanly. EOF } # ── Remote mine over MCP ───────────────────────────────────────────── # Usage: run_remote_mine <url> <token> <source> <wing> <agent> # run_remote_mine --self-test # # WHY THIS IS A FUNCTION WITH A SELF-TEST: MCP answers a hard tool failure with # HTTP 200 and a JSON-RPC *result* whose content[].text holds the tool's own # JSON as an ESCAPED STRING. This code used to decide success with # `'"error"' in body`, which can never match those bytes (they are \"error\"), # so on 2026-08-15 a mine that failed with # {"success": false, "error": "source directory not found: '/data/feed/...'"} # was reported as "Done. Wing updated." and nothing was filed. A silent # false success in a feeder is worse than a crash: the only artifact says it # worked. The fixtures below pin that exact body so it cannot come back. run_remote_mine() { python3 - "$@" <<'PY' import json, sys, urllib.error, urllib.request def classify(body): """Return (ok, note) for an MCP tools/call response body. ok=False means the mine demonstrably failed. note carries the reason, or — when ok is True — an "unverified" caveat if the response contained no JSON tool payload to adjudicate. Never claim more than the bytes support. """ try: env = json.loads(body) except ValueError: return False, "response was not JSON: " + body[:200].replace("\n", " ") if not isinstance(env, dict): return False, "response was not a JSON object" if env.get("error") is not None: # JSON-RPC transport-level error return False, "JSON-RPC error: " + json.dumps(env["error"])[:300] result = env.get("result") if not isinstance(result, dict): return False, "response carried no result object" if result.get("isError"): return False, "MCP isError set: " + json.dumps(result.get("content"))[:300] saw_payload = False for item in result.get("content") or []: text = item.get("text") if isinstance(item, dict) else None if not isinstance(text, str): continue try: payload = json.loads(text) # the escaped inner JSON except ValueError: continue # plain prose content: nothing to judge if not isinstance(payload, dict): continue saw_payload = True if payload.get("success") is False: return False, str(payload.get("error") or "tool reported success=false") if payload.get("error"): return False, str(payload["error"]) if not saw_payload: return True, "unverified: no JSON tool payload in the response" return True, "" FIXTURES = [ # 1. The real 2026-08-15 failure: HTTP 200, JSON-RPC result, tool failed. ('{"jsonrpc": "2.0", "id": 1, "result": {"content": [{"type": "text", ' '"text": "{\\n \\"success\\": false,\\n \\"error\\": \\"source directory ' 'not found: \'/data/feed/emb-7kj4vr4g\'\\"\\n}"}]}}', False), # 2. A real success. ('{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":' '"{\\"success\\": true, \\"mode\\": \\"convos\\", \\"output\\": \\"Drawers filed: 12\\"}"}]}}', True), # 3. JSON-RPC level error (bad method, auth rejected at protocol level). ('{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"Method not found"}}', False), # 4. MCP tool-level isError flag. ('{"jsonrpc":"2.0","id":1,"result":{"isError":true,"content":[{"type":"text","text":"boom"}]}}', False), # 5. Not JSON at all (proxy error page, 502 HTML). ('<html><body>502 Bad Gateway</body></html>', False), # 6. Success-shaped envelope with prose content: cannot be called a failure. ('{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"mined 3 files"}]}}', True), ] args = sys.argv[1:] if args[:1] == ["--self-test"]: failures = 0 for i, (body, want_ok) in enumerate(FIXTURES, 1): got_ok, note = classify(body) if got_ok != want_ok: failures += 1 print(f" [{'ok ' if got_ok == want_ok else 'FAIL'}] fixture {i}: " f"want_ok={want_ok} got_ok={got_ok} note={note[:70]!r}") # Regression guard: the detector this replaced must be shown blind to #1. old_detector_sees_it = '"error"' in FIXTURES[0][0] if old_detector_sees_it: failures += 1 print(f" [{'ok ' if not old_detector_sees_it else 'FAIL'}] regression guard: " f"substring detector sees fixture 1? {old_detector_sees_it} (must be False)") print("SELF-TEST " + ("FAILED" if failures else "PASSED")) sys.exit(1 if failures else 0) url, token, source, wing, agent = args[:5] payload = json.dumps({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "mempalace_mine", "arguments": {"source": source, "mode": "convos", "wing": wing, "agent": agent}, }, }).encode() headers = {"Content-Type": "application/json", "Accept": "application/json"} if token: headers["Authorization"] = f"Bearer {token}" try: with urllib.request.urlopen( urllib.request.Request(url, data=payload, headers=headers), timeout=900 ) as resp: body = resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace")[:500] if hasattr(exc, "read") else "" print(f"error: remote mine transport failed: HTTP {exc.code} {exc.reason} {detail}", file=sys.stderr) sys.exit(1) except urllib.error.URLError as exc: print(f"error: remote mine transport failed: {exc}", file=sys.stderr) sys.exit(1) print(body[:4000]) ok, note = classify(body) if not ok: print(f"error: remote mine reported failure: {note}", file=sys.stderr) sys.exit(1) if note: print(f"warning: remote mine {note}", file=sys.stderr) sys.exit(0) PY } # ── Parse args ─────────────────────────────────────────────────────── while [[ $# -gt 0 ]]; do case "$1" in -h|--help) usage; exit 0 ;; --wing) WING="${2:-}"; shift 2 ;; --session) SESSION_ID="${2:-}"; shift 2 ;; --since) SINCE="${2:-}"; shift 2 ;; --min-messages) MIN_MESSAGES="${2:-}"; shift 2 ;; --min-assistant-chars) MIN_ASSISTANT_CHARS="${2:-}"; shift 2 ;; --stage) STAGE_ROOT="${2:-}"; shift 2 ;; --mode) MODE="${2:-}"; shift 2 ;; --prepare) PREPARE_ONLY=1; shift ;; --reason) REASON="${2:-}"; shift 2 ;; --agent) AGENT="${2:-}"; shift 2 ;; --sessions-dir) PI_SESSIONS_DIR="${2:-}"; shift 2 ;; --dry-run) DRY_RUN=1; shift ;; --self-test) SELF_TEST=1; shift ;; --repair) DO_REPAIR=1; shift ;; --no-repair) shift ;; # deprecated alias; no-repair is the default --) shift; break ;; -*) echo "error: unknown option: $1" >&2; usage >&2; exit 1 ;; *) echo "error: unexpected arg: $1" >&2; exit 1 ;; esac done # The classifier self-test is pure logic: no palace, no network, no sessions # dir. Dispatch before preflight so it runs anywhere, including in CI. if [[ $SELF_TEST -eq 1 ]]; then echo "mempalace-pi-session --self-test: remote-mine response classifier" run_remote_mine --self-test exit $? fi # ── Preflight ──────────────────────────────────────────────────────── if [[ ! -d "$PI_SESSIONS_DIR" ]]; then echo "error: pi sessions dir not found at $PI_SESSIONS_DIR" >&2 echo " override with --sessions-dir <path> or PI_SESSIONS_DIR env var" >&2 exit 2 fi case "$MODE" in auto) if [[ -n "$REMOTE_URL" ]]; then MODE="remote"; else MODE="local"; fi ;; local|remote) ;; *) echo "error: --mode must be auto|local|remote" >&2; exit 1 ;; esac # The mempalace CLI is only needed when WE do the mine. --prepare never opens # the palace, and remote mode talks to the server over HTTP. if [[ $PREPARE_ONLY -eq 0 && "$MODE" == "local" ]] && ! command -v mempalace >/dev/null 2>&1; then echo "error: mempalace CLI not found in PATH" >&2 exit 3 fi if [[ "$MODE" == "remote" ]]; then command -v rsync >/dev/null 2>&1 || { echo "error: rsync not found (needed for --mode remote)" >&2; exit 3; } if [[ -z "$SSH_TARGET" ]]; then echo "error: MEMPALACE_PI_SSH_TARGET unset (needed for --mode remote)" >&2 exit 1 fi # The devbox generates a dedicated LAN-jump key/config; prefer it if present. if [[ -z "$SSH_CONFIG" && -f "$HOME/.ssh-local/config" ]]; then SSH_CONFIG="$HOME/.ssh-local/config" fi # Remote mode names the same inbox twice: where rsync PUTS the files, and # what the SERVER is told to mine. They may legitimately differ # (containerized server: host dir bind-mounted elsewhere), but when they # differ by accident rsync still succeeds and only the mine fails — the # 2026-08-15 /data/feed incident, where transcripts shipped for hours and # were filed nowhere. Warn in PREFLIGHT so --dry-run and --prepare see it # too, not just a full run that gets as far as shipping. SHIP_PATH="$SSH_TARGET" [[ "$SHIP_PATH" == *:* ]] && SHIP_PATH="${SHIP_PATH##*:}" if [[ "${SHIP_PATH%/}" != "${REMOTE_PATH%/}" ]]; then echo "note: shipping to '${SHIP_PATH%/}/$DEVICE' but asking the server to mine" echo " '${REMOTE_PATH%/}/$DEVICE'. Correct only if the palace server sees" echo " '${SHIP_PATH%/}' at '${REMOTE_PATH%/}' (containerized server with a bind" echo " mount). A NATIVE server (systemd unit / uv tool) sees host paths —" echo " then set MEMPALACE_PI_REMOTE_PATH='${SHIP_PATH%/}'." fi fi for _n in MIN_MESSAGES MIN_ASSISTANT_CHARS; do if ! [[ "${!_n}" =~ ^[0-9]+$ ]]; then _flag="--$(printf '%s' "${_n,,}" | tr '_' '-')" echo "error: $_flag must be an integer" >&2 exit 1 fi done # ── Staging dir ────────────────────────────────────────────────────── # Deterministic per-wing path so source_file dedup works across re-runs. See # "Staging location" in the header for why this should not be disposable. STAGE="${STAGE_ROOT%/}/$WING" mkdir -p "$STAGE" [[ -n "$REASON" ]] && echo "mempalace-pi-session [$REASON] mode=$MODE stage=$STAGE" # ── Single-writer guard ────────────────────────────────────────────── # Non-blocking: overlapping triggers (a session_shutdown landing on top of a # debounced mid-session run) must not queue or race. Losing a run is harmless # — the next one re-exports from scratch. exec 9>"${STAGE_ROOT%/}/.lock" if command -v flock >/dev/null 2>&1 && ! flock -n 9; then echo "another mempalace-pi-session run holds the lock; skipping" exit 0 fi # ── Export sessions (Python heredoc) ──────────────────────────────── # Parses pi JSONL files and writes Claude Code JSONL per session into $STAGE. # Also classifies each export as NEW/ALREADY FILED (by source_file lookup) # so --dry-run reports the real mine-set size. Classification is advisory; # `mempalace mine --mode convos` is still the authoritative dedup. export_count=$(python3 - "$PI_SESSIONS_DIR" "$STAGE" "$SESSION_ID" "$SINCE" "$MIN_MESSAGES" "$MIN_ASSISTANT_CHARS" "$MODE" <<'PY' import json, os, sqlite3, sys from datetime import datetime, timezone from pathlib import Path sessions_dir, stage, session_filter, since, min_messages, min_assistant_chars, mode = sys.argv[1:8] min_messages = int(min_messages) min_assistant_chars = int(min_assistant_chars) stage = Path(stage) sessions_dir = Path(sessions_dir) # Convert --since YYYY-MM-DD to epoch seconds (comparing against file mtime) since_epoch = None if since: try: since_epoch = datetime.strptime(since, "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp() except ValueError: print(f"error: --since must be YYYY-MM-DD, got {since!r}", file=sys.stderr) sys.exit(1) # ── Load palace's already-filed source_files (best-effort, read-only) ── # already_filed is None in remote mode: unknown, not empty. The dedup that # matters happens on the PALACE HOST and is keyed on the REMOTE inbox path, so # this machine's local palace file cannot answer the question — and answering # it anyway is how a preview comes to say "6 already filed" about a palace it # is not feeding (2026-08-15). An honest "[?]" beats a confident wrong number. already_filed = None if mode == "remote" else set() # Mirror mempalace's own resolution order (config.py): MEMPALACE_PALACE_PATH, # then the legacy MEMPAL_PALACE_PATH, then the default. NOT "MEMPALACE_PATH" — # that name is not a mempalace concept, and reading it silently degraded this # NEW/SKIP preview to "everything is new" wherever some other tool had set it. palace_path = ( os.environ.get("MEMPALACE_PALACE_PATH") or os.environ.get("MEMPAL_PALACE_PATH") or os.path.expanduser("~/.mempalace/palace") ) chroma_db = Path(palace_path) / "chroma.sqlite3" if already_filed is not None and chroma_db.is_file(): try: pcon = sqlite3.connect(f"file:{chroma_db}?mode=ro", uri=True) for (sf,) in pcon.execute( "SELECT DISTINCT string_value FROM embedding_metadata " "WHERE key='source_file' AND string_value LIKE ?", (f"{stage}%",), ): if sf: already_filed.add(sf) pcon.close() except sqlite3.Error: pass # palace unreachable → miner will dedup def extract_text(content): """Flatten a message content (string | list-of-blocks) to plain text. Drops image + thinking blocks; keeps text + renders toolCall/toolResult stubs inline. Returns ("", [tool_uses], [tool_results]) where tool_uses are collected for assistant messages and tool_results for toolResult messages. """ if isinstance(content, str): return content, [], [] if not isinstance(content, list): return "", [], [] text_parts = [] tool_uses = [] for block in content: if not isinstance(block, dict): continue bt = block.get("type") if bt == "text": t = block.get("text", "") if t: text_parts.append(t) elif bt == "thinking": # Drop reasoning content — high-noise, low-signal for search. continue elif bt == "image": # Palace is text-only. continue elif bt == "toolCall": tool_uses.append({ "type": "tool_use", "id": block.get("id") or "", "name": block.get("name") or "tool", "input": block.get("arguments") or {}, }) return "\n".join(text_parts), tool_uses, [] def load_session(path: Path): """Parse a pi JSONL session file. Returns (header, entries) or None.""" try: with path.open("r", encoding="utf-8") as f: lines = [ln for ln in f.read().splitlines() if ln.strip()] except OSError: return None if not lines: return None try: header = json.loads(lines[0]) except json.JSONDecodeError: return None if header.get("type") != "session": return None entries = [] for ln in lines[1:]: try: entries.append(json.loads(ln)) except json.JSONDecodeError: continue return header, entries def derive_title(entries, fallback: str) -> str: """Prefer session_info.name; else truncated first user message.""" # session_info entries: most-recent wins name = None for e in entries: if e.get("type") == "session_info" and e.get("name"): name = e["name"] if name: return name[:120] for e in entries: if e.get("type") != "message": continue msg = e.get("message") or {} if msg.get("role") != "user": continue text, _, _ = extract_text(msg.get("content")) text = " ".join(text.split()) # collapse whitespace if text: return (text[:80] + "…") if len(text) > 80 else text return fallback # Discover session files paths = sorted(sessions_dir.rglob("*.jsonl")) if session_filter: paths = [p for p in paths if session_filter in p.name] exported = 0 skipped_short = 0 skipped_quiet = 0 skipped_malformed = 0 skipped_already_filed = 0 for path in paths: try: mtime = path.stat().st_mtime except OSError: continue if since_epoch is not None and mtime < since_epoch: continue parsed = load_session(path) if parsed is None: skipped_malformed += 1 continue header, entries = parsed session_uuid = header.get("id") or path.stem cwd = header.get("cwd") or "?" header_ts = header.get("timestamp") or "" try: date_str = header_ts[:10] if header_ts else datetime.fromtimestamp( mtime, tz=timezone.utc).strftime("%Y-%m-%d") except Exception: date_str = datetime.fromtimestamp(mtime, tz=timezone.utc).strftime("%Y-%m-%d") # Count user+assistant message entries for the min-messages filter turn_count = sum( 1 for e in entries if e.get("type") == "message" and (e.get("message") or {}).get("role") in ("user", "assistant") ) if turn_count < min_messages: skipped_short += 1 continue title = derive_title(entries, fallback=session_uuid[:8]) assistant_chars = 0 out_lines = [] out_lines.append({ "type": "user", "message": { "content": f"[session: {title} | {cwd} | {date_str} | source: pi]" }, }) for e in entries: t = e.get("type") if t == "message": msg = e.get("message") or {} role = msg.get("role") if role == "user": text, _, _ = extract_text(msg.get("content")) if text.strip(): out_lines.append({"type": "user", "message": {"content": text}}) elif role == "assistant": text, tool_uses, _ = extract_text(msg.get("content")) assistant_chars += len(text.strip()) blocks = [] if text.strip(): blocks.append({"type": "text", "text": text}) blocks.extend(tool_uses) if not blocks: continue # Simplify single-text to string (matches mempalace-session). if len(blocks) == 1 and blocks[0].get("type") == "text": content = blocks[0]["text"] else: content = blocks out_lines.append({"type": "assistant", "message": {"content": content}}) elif role == "toolResult": text, _, _ = extract_text(msg.get("content")) tool_id = msg.get("toolCallId") or "" if not tool_id: continue out_lines.append({ "type": "human", "message": { "content": [{ "type": "tool_result", "tool_use_id": tool_id, "content": text or "(no output)", }], }, }) elif role == "bashExecution": # Rendered as a synthetic assistant annotation so the # command + output stay associated with the surrounding turn. cmd = msg.get("command") or "" out = msg.get("output") or "" exit_code = msg.get("exitCode") note = f"[user-bash] $ {cmd}\nexit={exit_code}\n{out}".strip() if note: out_lines.append({"type": "user", "message": {"content": note}}) elif role == "custom": if not msg.get("display"): continue text, _, _ = extract_text(msg.get("content")) if text.strip(): ctype = msg.get("customType") or "custom" out_lines.append({ "type": "user", "message": {"content": f"[custom:{ctype}] {text}"}, }) elif role in ("branchSummary", "compactionSummary"): summary = msg.get("summary") or "" if summary.strip(): out_lines.append({ "type": "user", "message": {"content": f"[{role}] {summary}"}, }) # thinking-only / empty messages silently dropped elif t in ( "model_change", "thinking_level_change", "compaction", "branch_summary", "label", "session_info", "custom", "custom_message", ): # Non-conversational entries: drop. (custom_message with # display=true could be included but we already get it via the # "custom" message role above when pi materializes one.) continue # Need at least 2 turns (header + one real turn) for the normalizer. if len(out_lines) < 2: skipped_short += 1 continue # Assistant *text* volume, tool results excluded: the signal that the # session actually did something, independent of how much injected # skill/context text inflated the user side. if assistant_chars < min_assistant_chars: skipped_quiet += 1 print( f" [QUIET] {path.name} ({turn_count} turns, {assistant_chars} assistant chars)", file=sys.stderr, ) continue out_path = stage / f"pi_{session_uuid}.jsonl" with out_path.open("w", encoding="utf-8") as f: for obj in out_lines: f.write(json.dumps(obj, ensure_ascii=False) + "\n") # Preserve session mtime on the staging file for dedup stability. try: os.utime(out_path, (mtime, mtime)) except OSError: pass exported += 1 if already_filed is None: is_filed = False # unknowable here; the palace host decides status = "? " else: is_filed = str(out_path) in already_filed if is_filed: skipped_already_filed += 1 status = "SKIP" if is_filed else "NEW " print(f" [{status}] {out_path.name} ({turn_count} turns)", file=sys.stderr) print(f"EXPORTED {exported}") print(f"ALREADY_FILED {-1 if already_filed is None else skipped_already_filed}") if skipped_short: print(f"SKIPPED_SHORT {skipped_short}", file=sys.stderr) if skipped_quiet: print(f"SKIPPED_QUIET {skipped_quiet}", file=sys.stderr) if skipped_malformed: print(f"SKIPPED_MALFORMED {skipped_malformed}", file=sys.stderr) PY ) # Parse counts from stdout count="$(printf '%s\n' "$export_count" | awk '/^EXPORTED / { print $2 }')" count="${count:-0}" already_filed="$(printf '%s\n' "$export_count" | awk '/^ALREADY_FILED / { print $2 }')" already_filed="${already_filed:-0}" # -1 means "unknown" (remote mode), so guard the arithmetic. if [[ "$already_filed" -lt 0 ]]; then to_file="$count"; else to_file=$(( count - already_filed )); fi if [[ "$count" -eq 0 ]]; then echo "no sessions qualified for export" exit 0 fi echo "" echo "Exported $count session(s) to $STAGE" if [[ "$already_filed" -lt 0 ]]; then # Remote mode: dedup lives on the palace host, keyed on the remote inbox # path. Do not translate "unknown" into a number. to_file="$count" echo " all $count shipped → the palace host dedups by source_file (remote mode:" echo " this machine cannot preview what it already holds)" else echo " $to_file new → will be filed on mine" echo " $already_filed already filed → will be skipped (dedup by source_file)" fi if [[ $DRY_RUN -eq 1 ]]; then echo "" if [[ "$already_filed" -lt 0 ]]; then echo "--dry-run: skipping ship+mine. A real run would ship $count session(s) to" echo " ${SSH_TARGET%/}/$DEVICE/ and let the palace host dedup them." elif [[ "$to_file" -eq 0 ]]; then echo "--dry-run: no new sessions to mine. A real run would skip all $count." else echo "--dry-run: skipping mine step. A real run would file $to_file new session(s)." fi exit 0 fi # ── Ship to the palace host (remote mode only) ─────────────────────── # mempalace_mine expands its source path in the SERVER process, so in remote # mode the exports have to physically exist over there. rsync --update is the # idempotent half; the mine is the other half. MINE_SOURCE="$STAGE" if [[ "$MODE" == "remote" ]]; then ssh_cmd="ssh" [[ -n "$SSH_CONFIG" ]] && ssh_cmd="ssh -F $SSH_CONFIG" echo "" echo "Shipping stage to ${SSH_TARGET%/}/$DEVICE/ ..." if ! rsync -a --update --no-owner --no-group \ -e "$ssh_cmd" \ --include='*.jsonl' --exclude='*' \ "$STAGE/" "${SSH_TARGET%/}/$DEVICE/"; then echo "error: rsync to ${SSH_TARGET%/}/$DEVICE/ failed" >&2 exit 5 fi MINE_SOURCE="${REMOTE_PATH%/}/$DEVICE" fi # ── Phase boundary ─────────────────────────────────────────────────── # --prepare hands the source path to the caller (the pi mempalace extension), # which mines it through the MCP client that already holds the palace. if [[ $PREPARE_ONLY -eq 1 ]]; then echo "" printf 'MINE_SOURCE=%s\n' "$MINE_SOURCE" exit 0 fi # ── Run the mine ───────────────────────────────────────────────────── echo "" echo "Mining into wing '$WING'..." if [[ "$MODE" == "remote" ]]; then if ! run_remote_mine "$REMOTE_URL" "$REMOTE_TOKEN" "$MINE_SOURCE" "$WING" "$AGENT"; then echo "error: remote mine failed" >&2 exit 5 fi else # Capture output so palace-level contention can be told apart from a real # failure. A live pi session holds the palace through its own mempalace-mcp, # and that session's extension mines what we just staged — so contention # means "already handled", not "broken". set +e mine_out="$(mempalace mine "$MINE_SOURCE" --mode convos --wing "$WING" --agent "$AGENT" 2>&1)" mine_rc=$? set -e printf '%s\n' "$mine_out" if [[ $mine_rc -ne 0 ]]; then if printf '%s' "$mine_out" | grep -q "is held by"; then echo "" echo "palace is held by a live session; it will mine the staged exports itself" exit 0 fi echo "error: mempalace mine failed" >&2 exit 4 fi fi # ── Repair index ───────────────────────────────────────────────────── if [[ $DO_REPAIR -eq 1 ]]; then echo "" echo "WARNING: --repair runs an in-place HNSW rebuild that has wiped" echo " live palaces on past runs. Proceeding in 3 seconds..." sleep 3 echo "Rebuilding HNSW index..." mempalace repair --yes fi echo "" echo "Done. Wing '$WING' updated. Remember to reconnect any live MCP sessions."