joakimp
a361b71c40
rsync --update skips a file whose mtime is not strictly newer than the receiver's. The stage file's mtime IS the source transcript's mtime (os.utime() at :903, "preserve session mtime for dedup stability"), so re-exporting a session that has not been appended to since its last ship produces a mtime that is not newer than what's already at the receiver — exactly the case a redactor upgrade needs to ship, because content differs while mtime does not. --update reports success and sends nothing. Reported and patched by pi@mbp-m1-2020 (evt_20260827T211925_9674a31da0b4, artifact art_20260827T211839_fa52af563105, sha256 f217e47e…), measured live: a scrubbed re-export of a dormant session (pi_01a03022-…f542) sat unshipped in the palace host's inbox while every local signal reported a clean stage, saved only because a host-side sweep happened to rewrite the remote copy independently that same day. --checksum compares content and ignores size/mtime entirely. Dropping --update outright was considered and rejected: rsync's default quick check already transfers on a SIZE difference alone, which is why the observed case (33-byte placeholder vs a 43-byte token) would have been masked as "fixed" by a change that only works until a redaction whose placeholder happens to match the secret's length. os.utime() at :903 is untouched — its backdating is a separate, load-bearing design call for dedup stability, out of scope for this fix. Added scripts/test-rsync-ship-idempotency.sh: ships a file, rewrites its content to an EQUAL-LENGTH string while restoring the original mtime (what os.utime() does), ships again, asserts the receiver's sha256 changed. Equal length is deliberate, not cosmetic — mismatched lengths would pass via the quick check alone and prove nothing about --checksum specifically; this is the same reasoning that ruled out dropping --update. Verified the test discriminates: passes against today's --checksum, fails against --update (checked by temporarily substituting the flag in a copy, not committed). Runs offline — a local rsync destination path exercises the same size/mtime/checksum comparison as the ssh transfer, no palace or network needed.
1056 lines
48 KiB
Bash
Executable File
1056 lines
48 KiB
Bash
Executable File
#!/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/<wing> (default
|
|
# <palace-root>/pi-stage/<wing> — 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 <n> (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=<path>, 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: <title> | <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 name defaults to "<harness>@<device>" when the device is known
|
|
# (MEMPALACE_PI_DEVICE is set on every enrolled devbox). That one string is the
|
|
# only place a palace write records BOTH which harness produced it and which
|
|
# machine it came from: mempalace core 3.7.1 stamps neither, and because every
|
|
# client shares one bearer token the server cannot tell them apart either. The
|
|
# per-device inbox path and the pi_*.jsonl filename encode the same two facts
|
|
# only incidentally, so anything filed outside the feeder had to be inferred.
|
|
# Falls back to $USER — which is what every drawer filed before this carries.
|
|
AGENT="${MEMPALACE_PI_DEVICE:+pi@${MEMPALACE_PI_DEVICE}}"
|
|
AGENT="${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. Defaults to
|
|
pi@$MEMPALACE_PI_DEVICE when that is set, else $USER.
|
|
The palace records neither the harness nor the machine
|
|
on a write, so this one string is what makes a drawer
|
|
attributable to both.
|
|
--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.
|
|
# The redactor is a sibling module, imported by the heredoc below. Exported
|
|
# rather than passed as argv so the argv unpack stays stable.
|
|
#
|
|
# ${BASH_SOURCE[0]} reports the path the script was INVOKED as, and the image
|
|
# installs /usr/local/bin/mempalace-pi-session as a symlink into
|
|
# /opt/mempalace-toolkit/bin. A naive dirname therefore yields /usr/local/bin,
|
|
# where the redactor does not exist — and because the import is fail-closed, that
|
|
# turns every feeder tick on every device into "refusing to stage". Measured: the
|
|
# symlinked invocation exited FATAL while the direct one worked, i.e. it would
|
|
# have stopped the whole fleet's memory feed at the next image bake. So chase the
|
|
# symlink chain, and offer fallbacks rather than betting on one answer.
|
|
#
|
|
# readlink -f is avoided deliberately: it is GNU/newer-BSD only, and this script
|
|
# also runs directly on macOS hosts.
|
|
_mp_resolve() {
|
|
local p="$1" target
|
|
while [ -L "$p" ]; do
|
|
target="$(readlink "$p")" || break
|
|
case "$target" in
|
|
/*) p="$target" ;;
|
|
*) p="$(dirname "$p")/$target" ;;
|
|
esac
|
|
done
|
|
printf '%s' "$p"
|
|
}
|
|
_mp_self="$(_mp_resolve "${BASH_SOURCE[0]}")"
|
|
MEMPALACE_REDACT_DIR="$(cd "$(dirname "$_mp_self")" && pwd):$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd):/opt/mempalace-toolkit/bin"
|
|
export MEMPALACE_REDACT_DIR
|
|
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
|
|
|
|
# ── Secret scrubbing before anything is staged ───────────────────────────────
|
|
# This is the last point at which a transcript is a plain in-memory value: the
|
|
# local mine reads the staged file, and the remote path rsyncs that same file
|
|
# byte-for-byte, so scrubbing here covers BOTH transports with one hook.
|
|
#
|
|
# FAIL CLOSED. If the redactor cannot be imported, staging is refused rather
|
|
# than done unscrubbed — a missing module means a broken install, and the whole
|
|
# point of this step is that a secret must not reach a shared palace. Override
|
|
# deliberately with MEMPALACE_FEED_ALLOW_UNSCRUBBED=1 if you ever need to feed a
|
|
# machine whose toolkit is older than this file.
|
|
# Colon-separated candidates: symlink-resolved dir, invoked dir, then the image's
|
|
# known install path. First one that has the module wins.
|
|
for _cand in os.environ.get("MEMPALACE_REDACT_DIR", "").split(":"):
|
|
if _cand and os.path.isdir(_cand):
|
|
sys.path.insert(0, _cand)
|
|
try:
|
|
import mempalace_redact as _redact
|
|
except Exception as _e: # noqa: BLE001 - any import failure is fatal by design
|
|
if os.environ.get("MEMPALACE_FEED_ALLOW_UNSCRUBBED", "").strip() in {"1", "true", "yes"}:
|
|
_redact = None
|
|
print(" [WARN] secret scrubber unavailable, staging UNSCRUBBED by request "
|
|
f"({_e})", file=sys.stderr)
|
|
else:
|
|
print(f" [FATAL] secret scrubber unavailable ({_e}); refusing to stage. "
|
|
"Set MEMPALACE_FEED_ALLOW_UNSCRUBBED=1 to override.", file=sys.stderr)
|
|
raise SystemExit(3)
|
|
|
|
_known = _redact.env_secrets() if _redact else []
|
|
_redactions = 0
|
|
_reported = 0
|
|
|
|
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
|
|
|
|
# Scrub the parsed objects, not the serialized text: string VALUES get
|
|
# rewritten while keys, ids and structure are left exactly as they are.
|
|
# (Scanning raw JSONL instead would also match escape artifacts like the
|
|
# "\\t" before a field name, inventing keys such as "tapiKey".)
|
|
if _redact is not None:
|
|
_found: list = []
|
|
out_lines = [_redact.scrub_obj(obj, _known, _found)[0] for obj in out_lines]
|
|
_hard = [x for x in _found if not x.rule.endswith("-reported")]
|
|
_soft = [x for x in _found if x.rule.endswith("-reported")]
|
|
_redactions += len(_hard)
|
|
_reported += len(_soft)
|
|
if _hard:
|
|
_by = {}
|
|
for x in _hard:
|
|
# "T2:github-pat" — rule says what matched, tier says how much to
|
|
# trust it, which is what the reader of this line actually needs.
|
|
_k = f"T{x.tier}:{x.rule}"
|
|
_by[_k] = _by.get(_k, 0) + 1
|
|
print(f" [REDACTED] {path.name} "
|
|
+ ", ".join(f"{k}={v}" for k, v in sorted(_by.items()))
|
|
+ " fp=" + ",".join(sorted({x.fingerprint for x in _hard})),
|
|
file=sys.stderr)
|
|
|
|
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}")
|
|
# Report the scrub outcome even when it is zero: "0 redactions" is a measurement,
|
|
# whereas printing nothing is indistinguishable from a scrubber that never ran.
|
|
if _redact is not None:
|
|
print(f" [scrub] {_redactions} redaction(s) applied, "
|
|
f"{_reported} name-anchored candidate(s) reported only "
|
|
f"(set MEMPALACE_REDACT_STRICT=1 to redact those too)", file=sys.stderr)
|
|
|
|
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. --checksum is the
|
|
# idempotent half; the mine is the other half.
|
|
#
|
|
# NOT --update: the stage file's mtime is deliberately the SOURCE transcript's
|
|
# mtime (see the os.utime() in the exporter, "preserve session mtime for dedup
|
|
# stability"), so a re-export of a session that has not been appended to since
|
|
# the last ship carries an mtime that is NOT newer than the receiver copy. With
|
|
# --update rsync then SKIPS it silently -- which is exactly the case that must
|
|
# ship after a redactor change, because the content differs while the mtime does
|
|
# not. Measured on mbp-m1-2020 2026-08-27: a scrubbed re-export of a dormant
|
|
# session was skipped and unscrubbed bytes stayed in the palace host's inbox,
|
|
# while every local signal reported a clean stage. --checksum compares content
|
|
# and keeps the ship idempotent without trusting timestamps.
|
|
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 --checksum --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."
|