self-directed-learning/scripts/ump_verify.py

#!/usr/bin/env python3
"""
ump_verify.py — verify that a record id persisted to the canonical UMP store.

Reads memory.ump.json directly from the file layer, independent of any
HTTP route. This is the "file-layer check" referenced in the
ump-write-verify-hook SKILL.md.

Why this exists:
  POST /ump/remember always returns {"id":"urn:ump:..."} on success —
  but the HTTP /ump/get route in @universalmemoryprotocol/core@0.1.0
  is unbound and returns {"error":"not_found"} for any id, including
  ones that did persist. So a caller who trusts the HTTP get for
  verification will always conclude the write failed, even when it
  succeeded. MCP tools have analogous issues with the recall pipeline
  returning empty for freshly-written records (Qdrant embedding lag).

The verification strategy: trust the FILE LAYER (which is what other
UMP-aware processes actually read), not the network surface.

Usage:
  ump_verify.py <urn>                     # verify one id
  ump_verify.py --list-last <N>           # last N records (for debugging)
  ump_verify.py --exists <urn>            # exit 0 if present, 3 if missing
  ump_verify.py --wait <urn> --timeout 5  # wait up to 5s for the id to land

Exit codes:
  0 — id present in canonical store (verified)
  2 — id NOT present (write did not persist; treat as silent drop)
  3 — id malformed (not a urn:ump:... string)
  4 — store file missing or unreadable
  5 — store file corrupted (not a JSON list)

Output: single-line JSON to stdout. Designed to be `eval $(ump_verify.py <urn>)`
or piped into downstream tooling.

The canonical store path comes from the UMP_DIR env var (matches what
ump-memory.service and the Hermes MCP ump tool use). Defaults to
/root/.hermes/state/ump-local/memory.ump.json if env is unset.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
from pathlib import Path

DEFAULT_STORE = "/root/.hermes/state/ump-local/memory.ump.json"
URN_RE = re.compile(r"^urn:ump:[a-z0-9]{52}$")


def resolve_store_path() -> Path:
    raw = os.environ.get("UMP_DIR", DEFAULT_STORE)
    p = Path(raw) / "memory.ump.json" if not raw.endswith(".json") else Path(raw)
    if not p.exists():
        # fall back to default if env points elsewhere
        if p != Path(DEFAULT_STORE) and Path(DEFAULT_STORE).exists():
            return Path(DEFAULT_STORE)
    return p


def load_records(path: Path) -> tuple[list, dict]:
    """Return (records, metadata). Metadata includes size + read errors."""
    if not path.exists():
        return [], {"error": "store_file_missing", "path": str(path)}
    try:
        data = json.loads(path.read_text())
    except json.JSONDecodeError as e:
        return [], {"error": "store_file_corrupted", "path": str(path), "detail": str(e)}
    if not isinstance(data, list):
        return [], {"error": "store_file_wrong_shape", "path": str(path),
                    "shape": type(data).__name__}
    return data, {"path": str(path), "size_bytes": path.stat().st_size, "record_count": len(data)}


def verify(id_to_find: str, store_path: Path) -> dict:
    """Find id in store. Returns the verify result dict."""
    if not URN_RE.match(id_to_find):
        return {
            "verified": False,
            "id": id_to_find,
            "reason": "malformed_urn",
            "expected_format": "urn:ump:<52 alphanumeric chars>",
        }
    records, meta = load_records(store_path)
    if "error" in meta:
        return {
            "verified": False,
            "id": id_to_find,
            "reason": meta["error"],
            "store": meta,
        }
    matches = [r for r in records if r.get("id") == id_to_find]
    if matches:
        r = matches[0]
        return {
            "verified": True,
            "id": id_to_find,
            "found": {
                "kind": r.get("kind"),
                "subject": (r.get("body") or {}).get("subject", "")[:120],
                "created": (r.get("time") or {}).get("created"),
                "owner": (r.get("scope") or {}).get("owner"),
                "integrity_hash": (r.get("integrity") or {}).get("content_hash"),
            },
            "store": meta,
        }
    return {
        "verified": False,
        "id": id_to_find,
        "reason": "not_in_store",
        "store": meta,
    }


def wait_for(id_to_find: str, timeout_s: float, store_path: Path) -> dict:
    """Poll the file until id appears or timeout. Roughly checks every 100ms."""
    deadline = time.time() + timeout_s
    last_result = verify(id_to_find, store_path)
    while time.time() < deadline:
        if last_result.get("verified"):
            last_result.setdefault("poll", {})
            last_result["poll"] = {"waited_s": round(timeout_s - (deadline - time.time()), 3)}
            return last_result
        time.sleep(0.1)
        last_result = verify(id_to_find, store_path)
    last_result["poll"] = {"waited_s": round(timeout_s, 3), "verified_after_wait": False}
    return last_result


def list_last(n: int, store_path: Path) -> dict:
    records, meta = load_records(store_path)
    if "error" in meta:
        return {"error": meta}
    return {
        "store": meta,
        "last_n": [
            {
                "id": r.get("id"),
                "kind": r.get("kind"),
                "subject": (r.get("body") or {}).get("subject", "")[:80],
                "created": (r.get("time") or {}).get("created"),
            }
            for r in records[-n:]
        ],
    }


def main() -> int:
    p = argparse.ArgumentParser(description="Verify a record is persisted to the canonical UMP store.")
    p.add_argument("id_to_verify", nargs="?", help="URN to verify (urn:ump:...)")
    p.add_argument("--store", help="Override store path (default: $UMP_DIR/memory.ump.json)")
    p.add_argument("--list-last", type=int, metavar="N", help="List last N records instead of verifying")
    p.add_argument("--exists", action="store_true", help="Exit 0 if present, exit 3 if missing (quiet)")
    p.add_argument("--wait", type=float, metavar="SECONDS", help="Poll up to SECONDS for the id to land")
    p.add_argument("--quiet", action="store_true", help="Only output on error (still exits non-zero)")
    args = p.parse_args()

    store = Path(args.store) if args.store else resolve_store_path()
    store = store if str(store).endswith(".json") else store / "memory.ump.json"

    if args.list_last is not None:
        out = list_last(args.list_last, store)
        print(json.dumps(out, indent=2))
        return 0 if "error" not in out else 5

    if not args.id_to_verify:
        p.error("Missing urn to verify (or pass --list-last N)")

    if args.wait:
        result = wait_for(args.id_to_verify, args.wait, store)
    else:
        result = verify(args.id_to_verify, store)

    if not args.quiet:
        print(json.dumps(result, indent=2))

    if not result.get("verified"):
        if result.get("reason") == "malformed_urn":
            return 3
        if "store_file" in (result.get("reason") or ""):
            return 4
        return 2

    if args.exists:
        # Re-output to match --exists quiet mode
        if args.quiet:
            print(f"{args.id_to_verify} verified")
    return 0


if __name__ == "__main__":
    sys.exit(main())