chorus/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 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 _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(rec: dict) -> str:
    """Re-issue one queued write idempotently. Returns 'ok' (landed/already-present),
    'drop' (permanent 4xx — can never succeed), or 'retry' (still offline / 5xx)."""
    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 the rest). Returns the number actually replayed. Safe to call when empty."""
    items = pending()
    if not items:
        return 0
    replayed = 0
    remaining: list[dict] = []
    for i, rec in enumerate(items):
        result = _replay(rec)
        if result == "ok":
            replayed += 1
        elif result == "drop":
            continue  # warned in _replay; remove from queue
        else:  # retry -> still offline; keep this and everything after, in order
            remaining = items[i:]
            break
    _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