mempalace-toolkit/bin/mempalace-pi-session

#!/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).
#
# 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

# ── 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
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.
  --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
}

# ── 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 ;;
    --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

# ── 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
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" <<'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 = sys.argv[1:7]
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 = 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 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
    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 {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}"
to_file=$(( count - already_filed ))

if [[ "$count" -eq 0 ]]; then
  echo "no sessions qualified for export"
  exit 0
fi

echo ""
echo "Exported $count session(s) to $STAGE"
echo "  $to_file new   → will be filed on mine"
echo "  $already_filed already filed → will be skipped (dedup by source_file)"

if [[ $DRY_RUN -eq 1 ]]; then
  if [[ "$to_file" -eq 0 ]]; then
    echo ""
    echo "--dry-run: no new sessions to mine. A real run would skip all $count."
  else
    echo ""
    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 ! python3 - "$REMOTE_URL" "$REMOTE_TOKEN" "$MINE_SOURCE" "$WING" "$AGENT" <<'PY'
import json, sys, urllib.error, urllib.request

url, token, source, wing, agent = sys.argv[1:6]
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.URLError as exc:
    print(f"error: remote mine transport failed: {exc}", file=sys.stderr)
    sys.exit(1)
print(body[:4000])
sys.exit(1 if '"error"' in body else 0)
PY
  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."