echo/echo-memory.plugin.src/skills/echo-memory/scripts/echo.py

#!/usr/bin/env python3
"""echo.py — the single validated, cross-platform client for the ECHO Obsidian
Local REST API.

Replaces the original echo.sh so the toolchain runs identically on Windows,
macOS, and Linux with no bash / GNU-vs-BSD `date` dependency — only a Python 3
interpreter (which the linter already requires).

Every read/write to the vault should go through this script. It centralizes:
auth injection, HTTP-status checking (non-zero exit on >=400 so a failed write
can never look like success), one bounded retry on 5xx/connection errors,
WHOLE-LINE idempotent append (a new line that is merely a substring of an
existing one is no longer wrongly skipped), correct `::` heading-target handling,
frontmatter field patches, a read-back-confirmed advisory multi-writer lock, a
one-call cold-start `load`, and scope show/set.

Network is keep-alive and connection-pooled (one persistent connection per thread,
reused across requests), and `read_many` fans bulk GETs across a thread pool — so
full-vault scripts (sweep, lint, recall rebuild) make a handful of warm round-trips
instead of hundreds of fresh TLS handshakes, and no longer blow past tool timeouts.

Config (owner/endpoint/key are machine-local — see echo_config; the plugin ships none):
  owner/endpoint/key   resolved by echo_config from ~/.claude/echo-memory/config.json
                       (env ECHO_OWNER / ECHO_BASE / ECHO_KEY override per field).
                       Set up a machine with `echo.py config init` then edit, or
                       `echo.py config set --owner ... --endpoint ... --key ...`.
  ECHO_VERIFY   default 1  — read-back verify after a PUT
  ECHO_LOCK_TTL default 900  — seconds before an advisory lock is considered stale
  ECHO_TIMEOUT  default 30  — per-request socket timeout (seconds)
  ECHO_WORKERS  default 8   — concurrency for read_many bulk reads
  ECHO_TODAY    YYYY-MM-DD to use as "today" (pass the conversation's currentDate)

Usage:
  echo.py load                                  # cold-start: the 6 orientation reads in one call
  echo.py get    <path>                         # print file contents (404 -> exit 44)
  echo.py map    <path>                         # document-map JSON (headings/blocks/frontmatter)
  echo.py ls     <dir>                          # directory listing JSON
  echo.py search <query...>                     # /search/simple
  echo.py put    <path> [file]                  # create/overwrite (body from file or stdin)
  echo.py post   <path> [file]                  # raw append (NON-idempotent; prefer `append`)
  echo.py append <path> <line>                  # idempotent append: skips only on an exact whole-line match
  echo.py patch  <path> <append|prepend|replace> <heading|frontmatter|block> <target> [file]
  echo.py fm     <path> <field> <json-value>    # create-or-replace a frontmatter scalar
  echo.py bump   <path> [YYYY-MM-DD]            # set frontmatter updated: to today (or given date)
  echo.py triage [--list --json | <proposals.json> [--apply]]   # one-tap inbox triage + audit log
  echo.py delete <path>                         # DELETE (destructive; explicit use only)
  echo.py lock   <owner-id>                     # acquire advisory lock (exit 75 if held & fresh, or lost the race)
  echo.py unlock <owner-id>                     # release advisory lock if owned by <owner-id>
  echo.py scope  show                           # print active scope, its freshness, and sessions-since
  echo.py scope  set "<text>"                   # switch scope atomically (history + replace + stamp scope_updated)
  echo.py config [show|init|set|import]         # machine-local owner/endpoint/key (config import <file> adopts a provided file)

Exit codes: 0 ok · 44 not-found(404) · 75 lock-held/lost · 76 duplicate-gate (capture) ·
            78 config-required · 2 usage · 1 other HTTP/transport error.
"""

from __future__ import annotations

import argparse
import datetime as dt
import http.client
import json
import locale
import os
import re
import sys
import tempfile
import threading
import time
import urllib.parse
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

# Never let a non-ASCII byte (em-dash in our own output, or Unicode in a fetched
# note) raise UnicodeEncodeError on a legacy Windows console.
for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8", errors="replace")  # type: ignore[union-attr]
    except Exception:
        pass

# Owner / endpoint / key are machine-local and resolved by echo_config from
# ~/.claude/echo-memory/config.json (env ECHO_OWNER / ECHO_BASE / ECHO_KEY override).
# The plugin ships none of them; load() returns "" for anything unset so --help,
# `config`, and `doctor` stay usable. Network ops raise a clear error if endpoint/key
# are missing (see _require_endpoint / request).
sys.path.insert(0, str(Path(__file__).resolve().parent))
import echo_config  # noqa: E402
_CFG = echo_config.load()
BASE = _CFG["endpoint"]
KEY = _CFG["key"]
OWNER = _CFG["owner"]
VERIFY = os.environ.get("ECHO_VERIFY", "1") != "0"
LOCK_TTL = int(os.environ.get("ECHO_LOCK_TTL", "900"))
# Per-request socket timeout. A single stuck file fails fast instead of stalling a
# full-vault pass; with read_many concurrency it only blocks one worker, not the run.
TIMEOUT = int(os.environ.get("ECHO_TIMEOUT", "30"))
# Concurrency for bulk reads (read_many). I/O-bound, so threads bypass the GIL fine.
MAX_WORKERS = max(1, int(os.environ.get("ECHO_WORKERS", "8")))

LOCK_PATH = "_agent/locks/vault.lock"


class EchoError(RuntimeError):
    def __init__(self, message: str, code: int = 1) -> None:
        super().__init__(message)
        self.code = code


def today() -> str:
    return os.environ.get("ECHO_TODAY") or dt.date.today().isoformat()


def now_iso() -> str:
    return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def vault_url(path: str) -> str:
    safe = "/".join(urllib.parse.quote(part) for part in path.split("/"))
    if path.endswith("/") and not safe.endswith("/"):
        safe += "/"
    return f"{BASE}/vault/{safe}"


# --- persistent connection pool (keep-alive) ---------------------------------
# urllib.urlopen opened a fresh TCP+TLS connection per call; against a remote HTTPS
# endpoint the repeated handshake dominated, so a full-vault pass made hundreds of
# serial handshakes and blew past the tool/sandbox timeout. We keep ONE connection
# per thread (http.client is not thread-safe, so each thread — main + every pool
# worker — owns its own) and reuse it across requests. read_many fans out over a
# thread pool, so the pool's connections stay warm for the duration of a sweep.
_local = threading.local()
MAX_ATTEMPTS = 3


def _require_config() -> None:
    """Fail loudly (not with an opaque DNS/connection error) when this machine has no
    usable key file yet — the most common fresh-install state. Also catches an
    untouched `config init` template (placeholder endpoint/key)."""
    if not echo_config.is_configured(_CFG):
        raise EchoError(
            "echo: NOT CONFIGURED — no usable ECHO key file on this machine. Ask the "
            f"operator for their config (owner/endpoint/key) and install it at "
            f"{echo_config.config_path()}: `echo.py config import <file>`, or "
            "`echo.py config set --owner ... --endpoint ... --key ...` "
            "(or set ECHO_BASE / ECHO_KEY).", 2)


def _new_connection() -> http.client.HTTPConnection:
    parts = urllib.parse.urlsplit(BASE)
    host = parts.hostname or parts.path
    if parts.scheme == "http":
        return http.client.HTTPConnection(host, parts.port or 80, timeout=TIMEOUT)
    return http.client.HTTPSConnection(host, parts.port or 443, timeout=TIMEOUT)


def _connection() -> http.client.HTTPConnection:
    conn = getattr(_local, "conn", None)
    if conn is None:
        conn = _new_connection()
        _local.conn = conn
    return conn


def _drop_connection() -> None:
    conn = getattr(_local, "conn", None)
    if conn is not None:
        try:
            conn.close()
        except Exception:
            pass
        _local.conn = None


def request(method: str, url: str, data: bytes | None = None,
            headers: dict[str, str] | None = None) -> tuple[int, bytes]:
    """Issue one request over this thread's reused connection. Returns (status, body);
    status 0 == transport failure (echo_queue relies on this to trigger offline queueing).
    Retries: a stale pooled connection reconnects immediately; 5xx and other transport
    errors get one bounded sleep-retry, same as before."""
    _require_config()
    merged = {"Authorization": f"Bearer {KEY}", **(headers or {})}
    parts = urllib.parse.urlsplit(url)
    target = parts.path or "/"
    if parts.query:
        target += "?" + parts.query
    last_status, last_body = 0, b""
    for attempt in range(MAX_ATTEMPTS):
        try:
            conn = _connection()
            conn.request(method, target, body=data, headers=merged)
            resp = conn.getresponse()
            body = resp.read()  # drain fully so the connection can be reused
            status = resp.status
            if status >= 500 and attempt < MAX_ATTEMPTS - 1:
                _drop_connection()
                time.sleep(1)
                continue
            return status, body
        except (http.client.RemoteDisconnected, http.client.BadStatusLine,
                ConnectionResetError, BrokenPipeError) as exc:
            # The server closed an idle keep-alive connection between requests.
            # Expected, not a failure: reconnect and retry without backing off.
            _drop_connection()
            last_status, last_body = 0, str(exc).encode()
            continue
        except Exception as exc:  # timeout, DNS, refused, TLS, ...
            _drop_connection()
            last_status, last_body = 0, str(exc).encode()
            if attempt < MAX_ATTEMPTS - 1:
                time.sleep(1)
                continue
    return last_status, last_body


def get_text(path: str) -> str | None:
    """GET a vault path -> decoded text, or None on 404/any error. Never raises, so a
    single unreadable note can't abort a bulk pass (resilience for sweep/lint)."""
    try:
        status, body = request("GET", vault_url(path))
    except Exception:
        return None
    return body.decode(errors="replace") if status == 200 else None


def read_many(paths) -> dict[str, str | None]:
    """Concurrently GET many vault paths -> {path: text_or_None}. The single biggest
    win for full-vault scripts: it turns hundreds of serial round-trips into MAX_WORKERS
    parallel ones over warm keep-alive connections. Resilient — a bad file maps to None."""
    unique = list(dict.fromkeys(paths))
    if not unique:
        return {}
    workers = min(MAX_WORKERS, len(unique))
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return dict(zip(unique, pool.map(get_text, unique)))


def check(status: int, body: bytes, context: str) -> None:
    if status == 404:
        raise EchoError(f"{context}: not found", 44)
    if status == 0:
        raise EchoError(f"{context}: vault unreachable ({body.decode(errors='replace')}) [{BASE}]", 1)
    if status >= 400:
        raise EchoError(f"{context}: HTTP {status} - {body.decode(errors='replace')}", 1)


def read_body(arg: str | None) -> bytes:
    if arg in (None, "-"):
        raw = sys.stdin.buffer.read()
        try:
            text = raw.decode("utf-8")
        except UnicodeDecodeError:
            text = raw.decode(locale.getpreferredencoding(False) or "cp1252", errors="replace")
        return text.encode("utf-8")
    return Path(arg).read_bytes()


def normalize_patch_body(data: bytes, op: str, target_type: str) -> bytes:
    """Heading PATCH bodies need a leading newline on replace and a trailing
    newline on append/prepend so the API does not concatenate onto adjacent text."""
    if target_type != "heading":
        return data
    text = data.decode("utf-8", errors="replace")
    if op == "replace":
        text = text.strip("\r\n")
        text = f"\n{text}\n" if text else "\n"
    elif text and not text.endswith("\n"):
        text += "\n"
    return text.encode("utf-8")


def normalize_json_scalar(value: str) -> bytes:
    """Accept either a raw scalar ('Fabrication Lead') or pre-formed JSON ('"2026-06-20"')."""
    value = value.strip()
    try:
        json.loads(value)
        return value.encode("utf-8")
    except json.JSONDecodeError:
        return json.dumps(value).encode("utf-8")


def temp_file(data: bytes) -> str:
    fh = tempfile.NamedTemporaryFile(delete=False)
    try:
        fh.write(data)
        return fh.name
    finally:
        fh.close()


# ---- commands ----------------------------------------------------------------

def cmd_get(path: str) -> int:
    status, body = request("GET", vault_url(path))
    check(status, body, f"get {path}")
    sys.stdout.buffer.write(body)
    return 0


def cmd_map(path: str) -> int:
    status, body = request("GET", vault_url(path),
                           headers={"Accept": "application/vnd.olrapi.document-map+json"})
    check(status, body, f"map {path}")
    sys.stdout.buffer.write(body)
    return 0


def cmd_ls(path: str) -> int:
    if not path.endswith("/"):
        path += "/"
    status, body = request("GET", vault_url(path))
    check(status, body, f"ls {path}")
    sys.stdout.buffer.write(body)
    return 0


def cmd_search(query: list[str]) -> int:
    q = urllib.parse.quote_plus(" ".join(query))
    status, body = request("POST", f"{BASE}/search/simple/?query={q}")
    check(status, body, "search")
    sys.stdout.buffer.write(body)
    return 0


def _qwrite(method: str, url: str, data: bytes | None = None, headers: dict | None = None):
    """A mutating request that survives an outage: routes through echo_queue.safe_request,
    which enqueues the write and returns (202, ...) when the vault is unreachable (H2)."""
    import echo_queue
    return echo_queue.safe_request(method, url, data=data, headers=headers)


def cmd_put(path: str, file_arg: str | None) -> int:
    data = read_body(file_arg)
    status, body = _qwrite("PUT", vault_url(path), data=data,
                           headers={"Content-Type": "text/markdown"})
    if status == 202:
        print(f"queued (offline): PUT {path} — will sync on the next reachable session")
        return 0
    check(status, body, f"put {path}")
    if VERIFY:
        status, _ = request("GET", vault_url(path))
        if status != 200:
            raise EchoError(f"put {path}: write did not verify (GET returned {status})")
    print(f"ok: PUT {path}")
    # Keep the recall (BM25) index current when a corpus note is written directly —
    # session logs, journal notes, hand-authored entity notes. update_note early-returns
    # for non-corpus paths and never raises, so this can't fail or slow a normal put.
    try:
        import echo_recall
        echo_recall.update_note(path, data.decode("utf-8", errors="replace"))
    except Exception as exc:  # noqa: BLE001 — index upkeep must never fail a put
        print(f"echo.py: recall-index update skipped ({exc})", file=sys.stderr)
    return 0


def cmd_post(path: str, file_arg: str | None) -> int:
    data = read_body(file_arg)
    status, body = _qwrite("POST", vault_url(path), data=data,
                           headers={"Content-Type": "text/markdown"})
    if status == 202:
        print(f"queued (offline): POST {path} — will sync on the next reachable session")
        return 0
    check(status, body, f"post {path}")
    print(f"ok: POST {path}")
    return 0


def cmd_append(path: str, line: str) -> int:
    """Idempotent append. Skips ONLY when the exact whole line already exists —
    a new line that is a substring of an existing one is still appended."""
    status, body = request("GET", vault_url(path))
    if status == 200:
        existing = [ln.rstrip("\r") for ln in body.decode(errors="replace").splitlines()]
        if line.rstrip("\r") in existing:
            print(f"skip: line already present in {path}")
            return 0
    elif status == 0:
        pass  # offline: skip the idempotency read; queue the POST (flush dedups by content)
    elif status != 404:
        check(status, body, f"append(read) {path}")
    status, body = _qwrite("POST", vault_url(path), data=f"{line}\n".encode(),
                           headers={"Content-Type": "text/markdown"})
    if status == 202:
        print(f"queued (offline): APPEND {path} — will sync on the next reachable session")
        return 0
    check(status, body, f"append {path}")
    print(f"ok: APPEND {path}")
    return 0


def cmd_patch(path: str, op: str, target_type: str, target: str, file_arg: str | None) -> int:
    if op not in {"append", "prepend", "replace"}:
        raise EchoError("patch: op must be append, prepend, or replace", 2)
    if target_type not in {"heading", "frontmatter", "block"}:
        raise EchoError("patch: target-type must be heading, frontmatter, or block", 2)
    data = normalize_patch_body(read_body(file_arg), op, target_type)
    status, body = _qwrite(
        "PATCH",
        vault_url(path),
        data=data,
        headers={
            "Operation": op,
            "Target-Type": target_type,
            "Target": target,
            "Content-Type": "application/json" if target_type == "frontmatter" else "text/markdown",
        },
    )
    if status == 202:
        print(f"queued (offline): PATCH {op} '{target}' -> {path} — will sync on the next reachable session")
        return 0
    check(status, body, f"patch {path} ({target})")
    print(f"ok: PATCH {op} {target_type} '{target}' -> {path}")
    return 0


def _yaml_flow(value) -> str:
    """Render a JSON-decoded value as a YAML flow scalar/list for one frontmatter line."""
    if isinstance(value, list):
        return "[" + ", ".join(_yaml_flow(v) for v in value) + "]"
    if isinstance(value, bool):
        return "true" if value else "false"
    if value is None:
        return ""
    s = str(value)
    return s if re.match(r"^[A-Za-z0-9][A-Za-z0-9 ._/-]*$", s) else json.dumps(s)


def _fm_upsert_line(text: str, field: str, rendered: str) -> str:
    """Surgically set `field: rendered` inside the YAML frontmatter block, creating the
    key (or the whole block) if absent. Every other line is preserved verbatim — this is
    the safety property that makes it OK to fall back from a failed PATCH."""
    line = f"{field}: {rendered}".rstrip()
    lines = text.splitlines()
    if lines and lines[0].strip() == "---":
        for end in range(1, len(lines)):
            if lines[end].strip() == "---":
                for i in range(1, end):  # key exists -> replace that one line
                    if re.match(rf"^{re.escape(field)}\s*:", lines[i]):
                        lines[i] = line
                        break
                else:
                    lines.insert(end, line)  # else append just before the closing ---
                return "\n".join(lines) + ("\n" if text.endswith("\n") else "")
    return f"---\n{line}\n---\n{text}"  # no frontmatter at all -> prepend a minimal block


def cmd_fm(path: str, field: str, value: str) -> int:
    """Create-or-replace a frontmatter scalar. PATCH replace handles the common case;
    a missing key returns 400 invalid-target (PATCH cannot create), which used to make
    `fm` fail on exactly the notes that needed repair — now it falls back to a surgical
    GET -> insert-one-line -> verified PUT."""
    data = normalize_json_scalar(value)
    try:
        return cmd_patch(path, "replace", "frontmatter", field, temp_file(data))
    except EchoError as exc:
        if "HTTP 400" not in str(exc):
            raise
    status, body = request("GET", vault_url(path))
    check(status, body, f"fm {path}")
    new = _fm_upsert_line(body.decode(errors="replace"), field,
                          _yaml_flow(json.loads(data.decode("utf-8"))))
    status, body = _qwrite("PUT", vault_url(path), data=new.encode("utf-8"),
                           headers={"Content-Type": "text/markdown"})
    if status == 202:
        print(f"queued (offline): FM {field} -> {path} — will sync on the next reachable session")
        return 0
    check(status, body, f"fm {path}")
    if VERIFY:
        status, body = request("GET", vault_url(path))
        if status != 200 or not re.search(rf"(?m)^{re.escape(field)}\s*:", body.decode(errors="replace")):
            raise EchoError(f"fm {path}: created key '{field}' did not verify on read-back")
    print(f"ok: FM created '{field}' in {path}")
    return 0


def cmd_delete(path: str) -> int:
    status, body = _qwrite("DELETE", vault_url(path))
    if status == 202:
        print(f"queued (offline): DELETE {path} — will sync on the next reachable session")
        return 0
    check(status, body, f"delete {path}")
    print(f"ok: DELETE {path}")
    return 0


def parse_lock_time(value: str) -> int:
    try:
        return int(dt.datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
                   .replace(tzinfo=dt.timezone.utc).timestamp())
    except Exception:
        # Unparseable timestamp -> treat the lock as FRESH (return "now"), so a
        # garbled lock is respected rather than silently stomped.
        return int(time.time())


def cmd_lock(owner: str, quiet: bool = False) -> int:
    status, body = request("GET", vault_url(LOCK_PATH))
    if status == 200 and body.strip():
        current = body.decode(errors="replace").strip()
        held_owner, _, held_iso = current.partition(" @ ")
        if held_owner != owner and int(time.time()) - parse_lock_time(held_iso) < LOCK_TTL:
            print(f"lock held by '{held_owner}' since {held_iso} (fresh)", file=sys.stderr)
            return 75
    status, body = request("PUT", vault_url(LOCK_PATH), data=f"{owner} @ {now_iso()}\n".encode(),
                           headers={"Content-Type": "text/markdown"})
    check(status, body, "lock")
    # Read-back confirm we actually hold it. The REST API has no compare-and-swap,
    # so two racers can both PUT; last writer wins. Whoever does not own the file
    # on read-back backs off instead of proceeding under a false lock.
    status, body = request("GET", vault_url(LOCK_PATH))
    if status == 200:
        held_owner = body.decode(errors="replace").split(" @ ", 1)[0].strip()
        if held_owner != owner:
            print(f"lock race: now held by '{held_owner}', not '{owner}' — backing off", file=sys.stderr)
            return 75
    if not quiet:
        print(f"ok: locked by {owner}")
    return 0


def cmd_unlock(owner: str, quiet: bool = False) -> int:
    status, body = request("GET", vault_url(LOCK_PATH))
    if status == 200:
        held_owner = body.decode(errors="replace").split(" @ ", 1)[0].strip()
        if held_owner != owner:
            print(f"lock owned by '{held_owner}', not '{owner}' — not releasing", file=sys.stderr)
            return 75
    status, body = request("DELETE", vault_url(LOCK_PATH))
    if status != 404:
        check(status, body, "unlock")
    if not quiet:
        print("ok: unlocked")
    return 0


def extract_heading(markdown: str, heading: str) -> str:
    """Return the body under a `## <heading>` up to the next `## ` heading."""
    out: list[str] = []
    capture = False
    marker = f"## {heading}"
    for line in markdown.splitlines():
        if line.strip() == marker:
            capture = True
            continue
        if capture and line.startswith("## "):
            break
        if capture:
            out.append(line)
    return "\n".join(out).strip()


def scope_show_op() -> dict:
    """Core scope-show (Phase 0): active scope + freshness + sessions-since, as data."""
    path = "_agent/context/current-context.md"
    status, body = request("GET", vault_url(path))
    check(status, body, "scope show")
    current = body.decode(errors="replace")
    scope_text = extract_heading(current, "Scope")
    scope_updated = ""
    for line in current.splitlines():
        if line.startswith("scope_updated:"):
            scope_updated = line.split(":", 1)[1].strip().strip('"').strip("'")
            break
    sessions_since = None
    # Count session logs dated after scope_updated (the drift signal).
    if scope_updated:
        status, body = request("GET", vault_url("_agent/sessions/"))
        if status == 200:
            try:
                files = json.loads(body).get("files", [])
                sessions_since = sum(1 for f in files if f.endswith(".md") and f[:10] > scope_updated)
            except json.JSONDecodeError:
                pass
    return {"ok": True, "action": "scope-show", "scope": scope_text,
            "scope_updated": scope_updated or None, "sessions_since": sessions_since}


def scope_set_op(text: str) -> dict:
    """Core scope-set (Phase 0): atomic switch (history + replace + stamp). Chatter
    from the underlying PATCHes routes to stderr; returns the switch envelope."""
    import contextlib
    if not text:
        raise EchoError("scope set needs the new scope text", 2)
    path = "_agent/context/current-context.md"
    status, body = request("GET", vault_url(path))
    check(status, body, "scope set")
    current = body.decode(errors="replace")
    prior = extract_heading(current, "Scope").replace("\n", " ").strip()[:140] or "(prior scope)"
    with contextlib.redirect_stdout(sys.stderr):
        cmd_patch(path, "prepend", "heading", "Current Context::Scope History",
                  temp_file(f"- {today()}: {prior}\n".encode()))
        cmd_patch(path, "replace", "heading", "Current Context::Scope",
                  temp_file(f"{text}\n".encode()))
        try:
            cmd_patch(path, "replace", "frontmatter", "scope_updated",
                      temp_file(json.dumps(today()).encode()))
        except EchoError as exc:
            raise EchoError(
                "scope set: body switched, but scope_updated frontmatter is missing "
                f"(run bootstrap.py to add it) [{exc}]"
            )
    return {"ok": True, "action": "scope-set", "scope": text, "prior": prior,
            "scope_updated": today()}


def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -> int:
    if subcommand == "show":
        env = scope_show_op()
        if as_json:
            print(json.dumps(env, ensure_ascii=False))
            return 0
        print("-- Active scope --")
        print(env["scope"])
        print(f"scope_updated: {env['scope_updated'] or '<missing — drift cannot be detected; run scope set or repair>'}")
        if env["sessions_since"] is not None:
            print(f"sessions logged since: {env['sessions_since']}")
        return 0
    if subcommand != "set":
        raise EchoError("scope: use 'show' or 'set \"<text>\"'", 2)
    env = scope_set_op(text or "")
    print(f"ok: scope switched (prior archived to Scope History; scope_updated={env['scope_updated']})")
    return 0


def _fetch_statuses(paths: list[str]) -> dict[str, tuple[int, bytes]]:
    """Concurrently GET vault paths keeping (status, body) per path — unlike read_many,
    the 404-vs-offline distinction survives, which cmd_load's cache logic needs."""
    if not paths:
        return {}
    workers = min(MAX_WORKERS, len(paths))
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return dict(zip(paths, pool.map(lambda p: request("GET", vault_url(p)), paths)))


def _bullet_lines(text: str) -> list[str]:
    return [ln for ln in text.splitlines() if ln.lstrip().startswith("- ")]


def _fm_line(text: str, field: str) -> str:
    for ln in text.splitlines():
        if ln.startswith(f"{field}:"):
            return ln.split(":", 1)[1].strip().strip('"').strip("'")
    return ""


def _render_brief(texts: dict[str, str | None], listing_files: list[str],
                  offline: bool) -> str:
    """The token-budgeted cold-start digest (2.1.0). Selection over compression: Fact /
    Pattern rules in full, recent Observations only, scope + freshness (not the whole
    history), the last session's key sections, today's Agent Log lines, inbox COUNT.
    Sections are trimmed lowest-priority-first when over ECHO_LOAD_BUDGET."""
    budget = int(os.environ.get("ECHO_LOAD_BUDGET", "8000"))
    S: list[tuple[str, str]] = []  # (section-name, text) in display order

    marker = texts.get("marker")
    if marker is None:
        S.append(("marker", "marker: _agent/echo-vault.md ABSENT — vault not bootstrapped "
                            "(run bootstrap.py before relying on memory)"))
    else:
        ver = _fm_line(marker, "schema_version") or "?"
        S.append(("marker", f"marker: _agent/echo-vault.md — bootstrapped, schema_version {ver}"))

    ctx = texts.get("context")
    if ctx:
        scope = extract_heading(ctx, "Scope") or "(empty)"
        upd = _fm_line(ctx, "scope_updated") or "unknown"
        since = sum(1 for f in listing_files if f.endswith(".md") and f[:10] > upd) \
            if upd != "unknown" else None
        head = f"scope (updated {upd}"
        if since is not None:
            head += f", {since} session(s) logged since"
        S.append(("scope", head + "):\n" + scope))
    else:
        S.append(("scope", "scope: current-context.md absent"))

    prefs = texts.get("preferences")
    if prefs:
        fact = extract_heading(prefs, "Fact / Pattern")
        if fact:
            S.append(("facts", "preferences — Fact / Pattern:\n" + fact))
        obs = _bullet_lines(extract_heading(prefs, "Observations"))
        if obs:
            shown = obs[-10:]
            head = f"preferences — Observations (last {len(shown)} of {len(obs)}):"
            S.append(("observations", head + "\n" + "\n".join(shown)))

    hb = texts.get("heartbeat")
    if hb and hb.strip():
        pointer = hb.strip().splitlines()[0]
        sess_path = pointer.split(" @ ", 1)[0].strip()
        part = [f"last session: {pointer}"]
        log_text = texts.get("_pointed_session")
        if log_text:
            for h in ("Goal", "Decisions Made", "Open Threads", "Suggested Next Step"):
                sec = extract_heading(log_text, h)
                if sec and sec.strip("() \n"):
                    part.append(f"  {h}: " + " / ".join(ln.strip() for ln in sec.splitlines()
                                                        if ln.strip())[:400])
        elif sess_path:
            part.append("  (session log not readable — see the path above)")
        S.append(("session", "\n".join(part)))
    elif listing_files:
        recent = sorted((f for f in listing_files if f.endswith(".md")), reverse=True)[:3]
        S.append(("session", "last session: heartbeat absent — recent logs:\n"
                  + "\n".join(f"  _agent/sessions/{f}" for f in recent)))

    today_note = texts.get("today")
    if today_note:
        log_lines = _bullet_lines(extract_heading(today_note, "Agent Log"))
        S.append(("agent-log", "today's Agent Log:\n" + ("\n".join(log_lines) or "  (empty)")))
    else:
        S.append(("agent-log", "today's Agent Log: (no daily note yet)"))

    inbox = texts.get("inbox")
    if inbox:
        items = re.findall(r"(?m)^\s*-\s*(\d{4}-\d{2}-\d{2})", inbox)
        if items:
            try:
                oldest = (dt.date.fromisoformat(today()) - dt.date.fromisoformat(min(items))).days
                S.append(("inbox", f"inbox: {len(items)} capture(s), oldest {oldest}d — "
                                   "offer triage if any are older than ~7 days"))
            except ValueError:
                S.append(("inbox", f"inbox: {len(items)} capture(s)"))
        else:
            S.append(("inbox", "inbox: empty"))
    else:
        S.append(("inbox", "inbox: empty"))

    def total() -> int:
        return sum(len(t) + 2 for _, t in S)

    # Over budget -> trim lowest-priority sections first, with explicit markers.
    for name, keep in (("observations", 4), ("agent-log", 4), ("session", 9)):
        if total() <= budget:
            break
        for i, (n, t) in enumerate(S):
            if n == name:
                lines = t.splitlines()
                if len(lines) > keep:
                    S[i] = (n, "\n".join(lines[:keep]) + "\n  (truncated — /echo-load for full)")
    out = f"ECHO load (brief) — {today()}\n\n" + "\n\n".join(t for _, t in S)
    if len(out) > budget:
        out = out[:budget] + "\n(truncated — /echo-load for full)"
    if offline:
        out += ("\n\nNOTE: vault unreachable — context above is last-known-good cache; "
                "writes this session will be queued and synced when the vault returns.")
    return out


LOAD_TARGETS = [
    ("marker",      "_agent/echo-vault.md"),
    ("preferences", "_agent/memory/semantic/operator-preferences.md"),
    ("context",     "_agent/context/current-context.md"),
    ("heartbeat",   "_agent/heartbeat/last-session.md"),
]


def _load_gather(targets):
    """Fetch the orientation reads (+ the sessions listing) in parallel and resolve
    each to text via status/cache. Shared by cmd_load and load_op. Returns
    (results, texts, listing_files, offline, marker_missing, heartbeat_absent)."""
    import echo_queue
    listing_path = "_agent/sessions/"
    results = _fetch_statuses([p for _, p in targets] + [listing_path])
    marker_missing = heartbeat_absent = offline = False
    texts: dict[str, str | None] = {}
    for label, path in targets:
        status, body = results[path]
        if status == 200:
            echo_queue.cache_put(path, body)  # refresh last-known-good for offline fallback
            texts[label] = body.decode(errors="replace")
        elif status == 0:  # vault unreachable -> degrade to last-known-good cache
            offline = True
            cached = echo_queue.cache_get(path)
            texts[label] = cached.decode(errors="replace") if cached is not None else None
        else:
            texts[label] = None
            if status == 404:
                if label == "marker":
                    marker_missing = True
                if label == "heartbeat":
                    heartbeat_absent = True
    lst_status, lst_body = results[listing_path]
    listing_files: list[str] = []
    if lst_status == 200:
        try:
            listing_files = list(json.loads(lst_body).get("files", []))
        except json.JSONDecodeError:
            pass
    return results, texts, listing_files, offline, marker_missing, heartbeat_absent


def _auto_fast_sweep(offline: bool) -> str | None:
    """Opportunistic index maintenance at load (2.3): when this machine's last fast
    sweep is older than ECHO_FAST_SWEEP_DAYS (default 7), run `sweep --fast` so the
    local recall + entity indexes true up against edits made by Obsidian or other
    clients. Best-effort; skipped offline; never raises."""
    if offline:
        return None
    try:
        import sweep as sweep_mod
        days = float(os.environ.get("ECHO_FAST_SWEEP_DAYS", "7"))
        age = sweep_mod.fast_sweep_age_days()
        if age is not None and age < days:
            return None
        return sweep_mod.fast()
    except Exception:  # noqa: BLE001 — maintenance must never block a load
        return None


def _fetch_pointed_session(texts: dict) -> None:
    """Follow the heartbeat pointer and stash the pointed session log for the digest."""
    hb = texts.get("heartbeat")
    if hb and hb.strip():
        sess_path = hb.strip().splitlines()[0].split(" @ ", 1)[0].strip()
        if sess_path:
            st2, b2 = request("GET", vault_url(sess_path))
            if st2 == 200:
                texts["_pointed_session"] = b2.decode(errors="replace")


def load_op(brief: bool = True) -> dict:
    """Core load (Phase 0): the orientation reads as data — sections keyed by label,
    plus offline/bootstrap flags, queue counts, recent sessions, and (with brief) the
    rendered digest. Raises EchoError(78) when the machine is not configured."""
    if not echo_config.is_configured(_CFG):
        raise EchoError("NOT CONFIGURED — no usable ECHO key file on this machine "
                        f"(expected at {echo_config.config_path()})", 78)
    import echo_queue
    synced = flagged = 0
    try:
        synced = echo_queue.flush()
        flagged = len(echo_queue.needs_attention())
    except Exception:  # noqa: BLE001 — queue upkeep must never block a load
        pass
    targets = LOAD_TARGETS + [("today", f"journal/daily/{today()}.md"),
                              ("inbox", "inbox/captures/inbox.md")]
    _, texts, listing_files, offline, marker_missing, _ = _load_gather(targets)
    swept = _auto_fast_sweep(offline)
    data = {"ok": True, "action": "load", "offline": offline,
            "marker_missing": marker_missing, "synced": synced,
            "needs_attention": flagged, "fast_sweep": swept,
            "sections": {k: v for k, v in texts.items() if not k.startswith("_")},
            "recent_sessions": sorted((f for f in listing_files if f.endswith(".md")),
                                      reverse=True)[:5]}
    if brief:
        _fetch_pointed_session(texts)
        data["brief"] = _render_brief(texts, listing_files, offline)
    return data


def cmd_load(brief: bool = False) -> int:
    """Cold-start orientation: the canonical 6 reads in one call (fetched in parallel).
    404s on today's daily note and the inbox are normal (printed as absent, not errors).
    `brief` renders the token-budgeted digest the SessionStart hook injects; the default
    full mode (and /echo-load) prints the raw sections unchanged."""
    if not echo_config.is_configured(_CFG):
        print("echo: NOT CONFIGURED — this machine has no usable ECHO key file yet.")
        print(f"Expected at: {echo_config.config_path()}")
        print("ACTION: ask the operator for their echo-memory config file (it holds the")
        print("        vault owner, endpoint, and API key), then install it with ONE of:")
        print("  python3 echo.py config import <path-to-their-file>")
        print("  python3 echo.py config set --owner \"…\" --endpoint \"https://…\" --key \"…\"")
        print("Then re-run load. (Memory is unavailable until configured.)")
        return 78  # distinct: configuration required
    targets = LOAD_TARGETS + [("today", f"journal/daily/{today()}.md"),
                              ("inbox", "inbox/captures/inbox.md")]
    import echo_queue
    # H2: sync any writes queued during a prior outage, best-effort and quiet on empty.
    try:
        synced = echo_queue.flush()
        if synced:
            print(f"(synced {synced} write(s) queued during a prior offline session)\n")
        flagged = echo_queue.needs_attention()
        if flagged:
            print(f"NOTE: {len(flagged)} queued write(s) need attention (e.g. a capture "
                  "stopped at the duplicate gate on replay) — resolve with capture "
                  "--merge-into/--force; `echo.py flush` re-lists them.\n")
    except Exception as exc:  # noqa: BLE001 — never let queue upkeep block a load
        print(f"(queue flush skipped: {exc})\n", file=sys.stderr)

    results, texts, listing_files, offline, marker_missing, heartbeat_absent = \
        _load_gather(targets)

    swept = _auto_fast_sweep(offline)
    if swept:
        print(f"({swept})\n")

    if brief:
        _fetch_pointed_session(texts)
        print(_render_brief(texts, listing_files, offline))
        return 0

    # ---- full mode: raw sections, output format unchanged ----
    for label, path in targets:
        status, body = results[path]
        if status == 200:
            print(f"===== {label}: {path} (HTTP 200) =====")
            text = texts[label] or ""
            sys.stdout.write(text)
            if not text.endswith("\n"):
                sys.stdout.write("\n")
        elif status == 404:
            print(f"===== {label}: {path} (HTTP 404) =====")
            print("(absent — fine)")
        elif status == 0:
            if texts[label] is not None:
                print(f"===== {label}: {path} (OFFLINE — serving stale cache) =====")
                sys.stdout.write(texts[label])
                if not texts[label].endswith("\n"):
                    sys.stdout.write("\n")
            else:
                print(f"===== {label}: {path} (OFFLINE — no cache) =====")
                print("(unreachable and not cached)")
        else:
            print(f"===== {label}: {path} (HTTP {status}) =====")
            print(f"(error HTTP {status}: {body.decode(errors='replace')[:200]})")
        print()
    # M3: heartbeat pointer missing/stale -> fall back to the recent sessions listing
    # (already fetched in the batch) so orientation works without the pointer.
    if heartbeat_absent and not offline and listing_files:
        files = sorted((f for f in listing_files if f.endswith(".md")), reverse=True)
        if files:
            print("===== recent sessions (heartbeat absent — fallback) =====")
            for f in files[:5]:
                print(f"  _agent/sessions/{f}")
            print()
    if offline:
        print("NOTE: vault unreachable — context above is last-known-good cache; writes this "
              "session will be queued and synced when the vault returns.")
    if marker_missing:
        print("NOTE: marker absent — vault not bootstrapped. Run bootstrap.py before relying on memory.")
    return 0


def cmd_config(args) -> int:
    """show | init | set the machine-local owner/endpoint/key config."""
    path = echo_config.config_path()
    if args.action == "init":
        p, created = echo_config.init_template()
        print(f"ok: wrote config template to {p} — edit it with your owner/endpoint/key"
              if created else f"config already exists at {p} (left unchanged)")
        return 0
    if args.action == "import":
        if not args.path:
            print("config import: pass the path to a config file "
                  "(JSON with owner/endpoint/key)", file=sys.stderr)
            return 2
        try:
            p = echo_config.import_file(args.path)
        except RuntimeError as exc:
            print(exc, file=sys.stderr)
            return 2
        print(f"ok: imported config to {p} (chmod 600 best-effort)")
        return 0
    if args.action == "set":
        if args.owner is None and args.endpoint is None and args.key is None:
            print("config set: pass at least one of --owner / --endpoint / --key", file=sys.stderr)
            return 2
        p = echo_config.write_config(args.owner, args.endpoint, args.key)
        print(f"ok: updated {p} (chmod 600 best-effort)")
        return 0
    # show — never prints the key in the clear
    cfg = echo_config.load()
    key = cfg["key"]
    redacted = (key[:4] + "…" + key[-4:]) if len(key) >= 12 else ("set" if key else "")
    print(f"config file: {path}" + ("" if path.exists() else " (absent)"))
    for field, shown in (("owner", cfg["owner"]), ("endpoint", cfg["endpoint"]), ("key", redacted)):
        src = echo_config.source(field)
        print(f"  {field:9} {shown or '(unset)'}   [{src}]")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Validated cross-platform ECHO vault client")
    sub = parser.add_subparsers(dest="cmd", required=True)
    p = sub.add_parser("load")
    p.add_argument("--brief", action="store_true")  # token-budgeted digest (hook default)
    sub.add_parser("flush")  # H2: replay writes queued during a prior outage
    sub.add_parser("doctor")  # M3: one-call readiness check
    sub.add_parser("get").add_argument("path")
    sub.add_parser("map").add_argument("path")
    sub.add_parser("ls").add_argument("path")
    sub.add_parser("search").add_argument("query", nargs="+")
    p = sub.add_parser("put"); p.add_argument("path"); p.add_argument("file", nargs="?")
    p = sub.add_parser("post"); p.add_argument("path"); p.add_argument("file", nargs="?")
    p = sub.add_parser("append"); p.add_argument("path"); p.add_argument("line")
    p = sub.add_parser("patch")
    p.add_argument("path"); p.add_argument("op"); p.add_argument("target_type")
    p.add_argument("target"); p.add_argument("file", nargs="?")
    p = sub.add_parser("fm"); p.add_argument("path"); p.add_argument("field"); p.add_argument("value")
    p = sub.add_parser("bump"); p.add_argument("path"); p.add_argument("date", nargs="?")
    sub.add_parser("delete").add_argument("path")
    sub.add_parser("lock").add_argument("owner")
    sub.add_parser("unlock").add_argument("owner")
    p = sub.add_parser("config")  # machine-local owner/endpoint/key (see echo_config)
    p.add_argument("action", nargs="?", default="show", choices=["show", "init", "set", "import"])
    p.add_argument("path", nargs="?")  # for `config import <path>`
    p.add_argument("--owner"); p.add_argument("--endpoint"); p.add_argument("--key")
    p = sub.add_parser("scope")
    p.add_argument("subcommand", nargs="?", default="show", choices=["show", "set"])
    p.add_argument("text", nargs="?")
    p.add_argument("--json", action="store_true")        # structured scope + freshness
    p = sub.add_parser("reflect")  # H5: apply a JSON proposal set (dry-run unless --apply)
    p.add_argument("file", nargs="?")
    p.add_argument("--apply", action="store_true")
    p = sub.add_parser("session-end")  # 2.1.0: one-call session end (log+agent-log+reflect+scope+heartbeat)
    p.add_argument("file", nargs="?")                    # bundle JSON; - or stdin
    p.add_argument("--apply", action="store_true")
    p = sub.add_parser("triage")   # one-tap inbox triage (reflect pipeline + audit log)
    p.add_argument("file", nargs="?")                    # proposals JSON; omit to list
    p.add_argument("--list", action="store_true", dest="list_inbox")
    p.add_argument("--json", action="store_true")        # structured inbox listing
    p.add_argument("--apply", action="store_true")       # route + write processing log

    # High-level operations (entity index + cross-links + recall). See echo_ops.py.
    sub.add_parser("resolve").add_argument("mention", nargs="+")
    p = sub.add_parser("recall"); p.add_argument("query", nargs="+"); p.add_argument("--limit", type=int, default=6)
    p.add_argument("--json", action="store_true")        # structured hits + neighbourhood
    p = sub.add_parser("link"); p.add_argument("a"); p.add_argument("b")
    p.add_argument("--json", action="store_true")
    p = sub.add_parser("capture")
    p.add_argument("title")
    p.add_argument("file", nargs="?")
    p.add_argument("--kind")
    p.add_argument("--inbox", action="store_true")
    p.add_argument("--status", default="")
    p.add_argument("--aliases", default="")
    p.add_argument("--source", default="")
    p.add_argument("--tags", default="")                  # extra tags; the kind is always seeded
    p.add_argument("--date")
    p.add_argument("--domain", default="business")
    p.add_argument("--no-log", action="store_true")
    p.add_argument("--json", action="store_true")       # M4: emit a JSON result envelope
    p.add_argument("--dry-run", action="store_true")     # M4: preview the plan, write nothing
    p.add_argument("--force", action="store_true")       # create despite a duplicate-gate hit
    p.add_argument("--merge-into", metavar="SLUG")       # route as an update to this entity
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        if args.cmd == "load":
            return cmd_load(brief=args.brief)
        if args.cmd == "doctor":
            import echo_doctor
            return echo_doctor.run()
        if args.cmd == "flush":
            import echo_queue
            n = echo_queue.flush()
            remaining = echo_queue.pending()
            flagged = [r for r in remaining if r.get("needs_attention")]
            if n:
                print(f"ok: flushed {n} queued write(s)"
                      + (f"; {len(remaining)} still queued" if remaining else ""))
            elif remaining:
                print(f"ok: {len(remaining)} write(s) still queued")
            else:
                print("ok: queue empty")
            for r in flagged:
                what = (f"capture '{(r.get('args') or {}).get('title')}'" if r.get("op") == "capture"
                        else f"{r.get('method')} {r.get('url')}")
                print(f"  needs attention ({r['needs_attention']}): {what} — resolve "
                      "(e.g. capture --merge-into <slug> / --force), it will not auto-land")
            return 0
        if args.cmd == "config":
            return cmd_config(args)
        if args.cmd == "get":
            return cmd_get(args.path)
        if args.cmd == "map":
            return cmd_map(args.path)
        if args.cmd == "ls":
            return cmd_ls(args.path)
        if args.cmd == "search":
            return cmd_search(args.query)
        if args.cmd == "put":
            return cmd_put(args.path, args.file)
        if args.cmd == "post":
            return cmd_post(args.path, args.file)
        if args.cmd == "append":
            return cmd_append(args.path, args.line)
        if args.cmd == "patch":
            return cmd_patch(args.path, args.op, args.target_type, args.target, args.file)
        if args.cmd == "fm":
            return cmd_fm(args.path, args.field, args.value)
        if args.cmd == "bump":
            return cmd_fm(args.path, "updated", json.dumps(args.date or today()))
        if args.cmd == "delete":
            return cmd_delete(args.path)
        if args.cmd == "lock":
            return cmd_lock(args.owner)
        if args.cmd == "unlock":
            return cmd_unlock(args.owner)
        if args.cmd == "scope":
            return cmd_scope(args.subcommand, args.text, as_json=args.json)
        if args.cmd == "reflect":
            import echo_reflect
            raw = read_body(args.file).decode("utf-8", errors="replace")
            try:
                proposals = json.loads(raw)
            except json.JSONDecodeError as exc:
                raise EchoError(f"reflect: proposals must be a JSON array ({exc})", 2)
            if not isinstance(proposals, list):
                raise EchoError("reflect: proposals must be a JSON array of objects", 2)
            return echo_reflect.apply(proposals, confirm=args.apply)
        if args.cmd == "session-end":
            import echo_session
            raw = read_body(args.file).decode("utf-8", errors="replace")
            try:
                bundle = json.loads(raw)
            except json.JSONDecodeError as exc:
                raise EchoError(f"session-end: bundle must be a JSON object ({exc})", 2)
            return echo_session.session_end(bundle, apply=args.apply)
        if args.cmd == "triage":
            import echo_triage
            if args.list_inbox or not args.file:
                return echo_triage.list_inbox(as_json=args.json)
            raw = read_body(args.file).decode("utf-8", errors="replace")
            try:
                proposals = json.loads(raw)
            except json.JSONDecodeError as exc:
                raise EchoError(f"triage: proposals must be a JSON array ({exc})", 2)
            if not isinstance(proposals, list):
                raise EchoError("triage: proposals must be a JSON array of objects", 2)
            return echo_triage.apply(proposals, confirm=args.apply)
        if args.cmd in ("resolve", "recall", "link", "capture"):
            import echo_ops  # lazy: only the high-level ops pull in the index/link modules
            if args.cmd == "resolve":
                return echo_ops.resolve(" ".join(args.mention))
            if args.cmd == "recall":
                return echo_ops.recall(args.query, limit=args.limit, as_json=args.json)
            if args.cmd == "link":
                return echo_ops.link(args.a, args.b, as_json=args.json)
            return echo_ops.capture(
                args.kind, args.title, args.file, status_v=args.status,
                aliases=args.aliases.split(",") if args.aliases else [],
                sources=args.source.split(",") if args.source else [],
                tags=args.tags.split(",") if args.tags else [],
                date=args.date, domain=args.domain, inbox=args.inbox, no_log=args.no_log,
                as_json=args.json, dry_run=args.dry_run,
                force=args.force, merge_into=args.merge_into)
    except RuntimeError as exc:
        # Catches EchoError from THIS module and from helper modules' own `import echo`
        # twin (when echo.py runs as __main__ the two class objects differ, but both
        # subclass RuntimeError and carry .code).
        print(f"echo.py: {exc}", file=sys.stderr)
        return getattr(exc, "code", 1)
    return 2


if __name__ == "__main__":
    raise SystemExit(main())