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

#!/usr/bin/env python3
"""echo_queue.py — H2: offline write-ahead queue + read cache.  [v1.0 — wired]

Today vault-unreachable => "proceed without memory" (SKILL.md), so a 502 (Obsidian not
running — the most common real failure) loses everything said that session. This module
makes explicit writes durable across an outage and lets cold-start `load` degrade to
last-known context instead of nothing.

DESIGN:
  * Outbox  : append-only NDJSON of intended writes. Replayed in order on the next
              reachable session. Replay is idempotent by construction — PUT/DELETE/PATCH-
              replace overwrite, and POST / PATCH-append are content-deduped (skip if the
              line/block is already present), the same strategy echo.py append uses. Each
              record carries an idem_key so the SAME write is never queued twice.
  * Cache   : last-known-good GET bodies, so `load` has a fallback. Written by cmd_load.
  * Location: a LOCAL state dir — it CANNOT live in the vault, because the vault is the
              thing that's down. Default ~/.echo-memory/, override ECHO_STATE_DIR.

Integration seam: `safe_request()` — the write verbs (cmd_put/post/append/patch/delete)
call it instead of echo.request; on an outage it enqueues and returns a synthesized
(202, b"queued ...") so the caller reports "queued, will sync" instead of failing.
"""
from __future__ import annotations

import base64
import hashlib
import json
import os
import sys
import urllib.parse
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import echo  # noqa: E402

MUTATING = {"PUT", "POST", "PATCH", "DELETE"}


def state_dir() -> Path:
    return Path(os.environ.get("ECHO_STATE_DIR") or (Path.home() / ".echo-memory"))


def outbox_path() -> Path:
    return state_dir() / "outbox.ndjson"


def cache_dir() -> Path:
    return state_dir() / "cache"


def _ensure_dirs() -> None:
    state_dir().mkdir(parents=True, exist_ok=True)
    cache_dir().mkdir(parents=True, exist_ok=True)


def _key(method: str, url: str, data: bytes | None) -> str:
    h = hashlib.sha1()
    h.update(method.encode())
    h.update(b"\0")
    h.update(url.encode())
    h.update(b"\0")
    h.update(data or b"")
    return h.hexdigest()


# --------------------------------------------------------------------- outbox
def enqueue(method: str, url: str, data: bytes | None, headers: dict, idem_key: str | None = None) -> None:
    """Append one intended write to the outbox (NDJSON; base64 body for binary safety).
    Dedups: if a record with the same idem_key is already pending, do nothing."""
    idem_key = idem_key or _key(method, url, data)
    if any(rec.get("idem_key") == idem_key for rec in pending()):
        return
    _ensure_dirs()
    rec = {
        "method": method.upper(), "url": url,
        "body_b64": base64.b64encode(data).decode() if data else None,
        "headers": headers or {}, "idem_key": idem_key,
        # No wall-clock: replay order is file order, not a timestamp (keeps this testable).
    }
    with outbox_path().open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(rec, ensure_ascii=False) + "\n")


def enqueue_capture(args: dict) -> None:
    """Queue a whole capture as ONE semantic record (2.1.0). Byte-level replay is wrong
    for the routed capture path: the create-vs-update decision, duplicate gate, and
    aliasing must re-run against the index AS IT IS AT REPLAY TIME, so flush re-invokes
    echo_ops.capture with the recorded arguments instead of replaying stale requests.
    `args` carries body_text inline (never a temp-file path) plus `today` (the capture-
    time ECHO_TODAY) so replayed frontmatter/log dates reflect when it was said."""
    key = "capture:" + hashlib.sha1(
        json.dumps(args, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
    if any(rec.get("idem_key") == key for rec in pending()):
        return
    _ensure_dirs()
    rec = {"op": "capture", "args": args, "idem_key": key}
    with outbox_path().open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(rec, ensure_ascii=False) + "\n")


def pending() -> list[dict]:
    p = outbox_path()
    if not p.exists():
        return []
    return [json.loads(ln) for ln in p.read_text(encoding="utf-8").splitlines() if ln.strip()]


def needs_attention() -> list[dict]:
    """Queued records that replay could not land automatically (e.g. a duplicate-gate
    stop) — surfaced by `flush` and `load` so the operator resolves them deliberately."""
    return [rec for rec in pending() if rec.get("needs_attention")]


def _rewrite(records: list[dict]) -> None:
    p = outbox_path()
    if not records:
        if p.exists():
            p.unlink()
        return
    _ensure_dirs()
    tmp = p.with_suffix(".ndjson.tmp")
    tmp.write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in records), encoding="utf-8")
    os.replace(tmp, p)  # atomic swap so a crash mid-flush can't corrupt the queue


def _rebase(url: str) -> str:
    """Re-point a queued URL at the CURRENT echo.BASE. Writes are queued with the base
    that was active when they failed; on replay the vault may be back at a different base
    (e.g. an endpoint migration), so we always reconstruct the origin from echo.BASE."""
    parts = urllib.parse.urlsplit(url)
    return echo.BASE + parts.path + (("?" + parts.query) if parts.query else "")


def _replay_capture(rec: dict) -> str:
    """Replay a queued semantic capture through echo_ops.capture against the CURRENT
    index. Returns 'ok', 'retry' (still offline), or 'gated' (duplicate gate / needs
    the operator — the record is kept and flagged, later records still replay)."""
    # Cheap reachability probe FIRST: if the vault is still down, echo_ops.capture's
    # own offline short-circuit would try to re-enqueue this very record — probe and
    # bail as 'retry' instead of recursing into the queue.
    st, _ = echo.request("GET", echo.vault_url("_agent/echo-vault.md"))
    if st == 0:
        return "retry"
    import echo_ops
    a = dict(rec.get("args") or {})
    body_text = a.pop("body_text", "") or ""
    day = a.pop("today", None)
    bodyfile = echo.temp_file(body_text.encode("utf-8")) if body_text else None
    prev = os.environ.get("ECHO_TODAY")
    try:
        if day:
            os.environ["ECHO_TODAY"] = day  # replayed dates reflect when it was said
        rc = echo_ops.capture(
            a.get("kind"), a.get("title") or "(untitled)", bodyfile,
            status_v=a.get("status_v", ""), aliases=a.get("aliases") or [],
            sources=a.get("sources") or [], tags=a.get("tags") or [],
            date=a.get("date"), domain=a.get("domain", "business"),
            inbox=bool(a.get("inbox")), no_log=bool(a.get("no_log")),
            force=bool(a.get("force")), merge_into=a.get("merge_into"))
    except Exception as exc:  # noqa: BLE001 — a broken record must not wedge the queue
        print(f"echo_queue: queued capture '{a.get('title')}' failed on replay ({exc}) "
              "— kept and flagged", file=sys.stderr)
        return "gated"
    finally:
        if day:
            if prev is None:
                os.environ.pop("ECHO_TODAY", None)
            else:
                os.environ["ECHO_TODAY"] = prev
    if rc == 0:
        return "ok"
    if rc == 76:
        print(f"echo_queue: queued capture '{a.get('title')}' stopped at the duplicate "
              "gate on replay — resolve with capture --merge-into <slug> or --force, "
              "then remove it from the outbox", file=sys.stderr)
    return "gated"


def _replay(rec: dict) -> str:
    """Re-issue one queued write idempotently. Returns 'ok' (landed/already-present),
    'drop' (permanent 4xx — can never succeed), 'gated' (kept + flagged for the
    operator), or 'retry' (still offline / 5xx)."""
    if rec.get("op") == "capture":
        return _replay_capture(rec)
    method = rec["method"]
    url = _rebase(rec["url"])
    data = base64.b64decode(rec["body_b64"]) if rec.get("body_b64") else None
    headers = rec.get("headers") or {}

    # Append-style writes: skip if the content is already present (idempotency on replay).
    is_append = method == "POST" or (method == "PATCH" and headers.get("Operation") == "append")
    if is_append and data:
        st, cur = echo.request("GET", url)
        if st == 0:
            return "retry"  # still offline
        if st == 200 and data.decode("utf-8", "replace").strip("\n") in cur.decode("utf-8", "replace"):
            return "ok"     # already landed (or a duplicate we must not re-add)

    st, body = echo.request(method, url, data=data, headers=headers)
    if st == 0 or 500 <= st < 600:
        return "retry"
    if 400 <= st < 500:
        print(f"echo_queue: dropping a queued {method} {url} — permanent HTTP {st} on replay",
              file=sys.stderr)
        return "drop"
    return "ok"


def flush() -> int:
    """Replay queued writes in order; stop at the first still-unreachable one (keeping it
    and everything after). A 'gated' record (duplicate gate / replay error) is kept and
    flagged but does NOT block later records — they are independent writes. Returns the
    number actually replayed. Safe to call when empty."""
    items = pending()
    if not items:
        return 0
    replayed = 0
    remaining: list[dict] = []
    stopped = False
    for rec in items:
        if stopped:
            remaining.append(rec)
            continue
        result = _replay(rec)
        if result == "ok":
            replayed += 1
        elif result == "drop":
            continue  # warned in _replay; remove from queue
        elif result == "gated":
            rec["needs_attention"] = rec.get("needs_attention") or "replay-gated"
            remaining.append(rec)
        else:  # retry -> still offline; keep this and everything after, in order
            remaining.append(rec)
            stopped = True
    _rewrite(remaining)
    return replayed


# --------------------------------------------------------------------- cache
def cache_key(path: str) -> Path:
    return cache_dir() / (path.replace("/", "__") + ".cache")


def cache_put(path: str, body: bytes) -> None:
    _ensure_dirs()
    cache_key(path).write_bytes(body)


def cache_get(path: str) -> bytes | None:
    p = cache_key(path)
    return p.read_bytes() if p.exists() else None


# ----------------------------------------------------------- integration seam
def safe_request(method: str, url: str, data=None, headers=None, idem_key: str | None = None):
    """Network-first wrapper around echo.request with offline queueing for writes.

    - success or a client error (4xx): return (status, body) as-is — 4xx is a bug, fail loud.
    - transport failure (status 0) or 5xx (after echo.request's own retry) on a MUTATING
      verb: enqueue the write and return (202, b"queued (offline)") so the caller reports it
      as durably queued rather than failed.
    - otherwise (e.g. a failing GET): return (status, body); the read-cache fallback lives
      in cmd_load, which knows which reads are worth serving stale.
    """
    method = method.upper()
    status, body = echo.request(method, url, data=data, headers=headers)
    offline = status == 0 or 500 <= status < 600
    if offline and method in MUTATING:
        enqueue(method, url, data, headers or {}, idem_key)
        return 202, b"queued (offline)"
    return status, body