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

#!/usr/bin/env python3
"""test_v1_scaffold.py — H4 interface guard for the v1.0 scaffolds.  [offline, no creds]

Asserts the scaffold contract holds so the skeleton can't silently rot while 1.0 is
built: every planned module imports, the parts that ARE implemented work (BM25 ranking,
queue/cache round-trip, slug helpers, secret resolution order), and the parts that are
TODO raise NotImplementedError (so a stub can't masquerade as done).

Run:  python test_v1_scaffold.py
"""
from __future__ import annotations

import os
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

failures: list[str] = []


def check(name: str, cond: bool, detail: str = "") -> None:
    print(f"{'ok  ' if cond else 'FAIL'} {name}" + (f"  -- {detail}" if not cond else ""))
    if not cond:
        failures.append(name)


def main() -> int:
    # H1 — BM25 core is implemented; assert it actually ranks a relevant doc first.
    import echo_recall as r
    ix = r.Bm25Index()
    ix.add("notes/mqtt.md", "the mqtt broker config and tls certificate rotation")
    ix.add("notes/payroll.md", "quarterly payroll run and employee deductions")
    top = ix.score("certificate rotation broker", limit=2)
    check("H1 BM25 ranks relevant doc first", bool(top) and top[0][0] == "notes/mqtt.md", str(top))
    check("H1 BM25 survives json round-trip",
          r.Bm25Index.from_json(ix.to_json()).score("payroll")[0][0] == "notes/payroll.md")
    ix.remove("notes/payroll.md")
    check("H1 BM25 remove() drops a doc's postings", ix.score("payroll") == [])
    check("H1 recall path is wired (not a stub)",
          all(callable(getattr(r, n, None))
              for n in ("recall", "rebuild", "update_note", "load_index", "save_index")))

    # H2 — cache round-trips, enqueue dedups, flush is a no-op on an empty queue (offline-safe).
    import echo_queue as q
    with tempfile.TemporaryDirectory() as d:
        os.environ["ECHO_STATE_DIR"] = d
        q.cache_put("a/b.md", b"hello")
        check("H2 cache round-trip", q.cache_get("a/b.md") == b"hello")
        q.enqueue("PUT", "http://x/vault/a.md", b"body", {}, idem_key="k1")
        check("H2 enqueue persists a record", len(q.pending()) == 1)
        q.enqueue("PUT", "http://x/vault/a.md", b"body", {}, idem_key="k1")
        check("H2 enqueue dedups by idem_key", len(q.pending()) == 1)
    with tempfile.TemporaryDirectory() as d2:
        os.environ["ECHO_STATE_DIR"] = d2
        check("H2 flush() returns 0 on an empty queue (no network)", q.flush() == 0)

    # H3 — lock CM + atomic index update are stubs.
    import echo_concurrency as c
    check("H3 auto_owner is stable", c.auto_owner() == c.auto_owner())
    check("H3 vault_lock returns a context manager (no network until entered)",
          hasattr(c.vault_lock(), "__enter__") and hasattr(c.vault_lock(), "__exit__"))
    check("H3 atomic_index_update is wired", callable(c.atomic_index_update))

    # H5 — proposal validation is pure (offline): keeps valid (incl. inbox), drops the rest.
    import echo_reflect as rf
    v, errs = rf.validate([
        {"title": "Ok", "kind": "person", "confidence": 0.9},
        {"title": "", "kind": "person"},                          # missing title
        {"title": "Bad", "kind": "bogus"},                        # unknown kind
        {"title": "Weak", "kind": "person", "confidence": 0.2},   # below confidence floor
        {"title": "Note", "inbox": True},                         # inbox needs no kind
    ])
    check("H5 validate keeps valid (incl. inbox), drops invalid", len(v) == 2 and len(errs) == 3, (len(v), errs))
    check("H5 classify/preview/apply wired",
          all(callable(getattr(rf, n, None)) for n in ("classify", "preview", "apply")))

    # Config — owner/endpoint/key resolve from env -> machine-local config.json (no baked
    # defaults). Env wins per field; the file fills the rest; missing required -> raises.
    import echo_config as cfg
    saved_env = {k: os.environ.get(k) for k in ("ECHO_KEY", "ECHO_BASE", "ECHO_OWNER", "ECHO_CONFIG")}
    try:
        for k in saved_env:
            os.environ.pop(k, None)
        with tempfile.TemporaryDirectory() as d:
            cfgfile = os.path.join(d, "config.json")
            os.environ["ECHO_CONFIG"] = cfgfile
            # init writes the placeholder template only when absent.
            p, created = cfg.init_template()
            check("config init scaffolds the template", created and os.path.exists(cfgfile))
            p2, created2 = cfg.init_template()
            check("config init is idempotent (no clobber)", created2 is False)
            # write_config persists a filled-in config; load() reads it back.
            cfg.write_config("Owner Name", "https://obsidian.example.com/", "file-key-xyz")
            loaded = cfg.load()
            check("config file round-trips owner/endpoint/key",
                  loaded == {"owner": "Owner Name",
                             "endpoint": "https://obsidian.example.com",  # trailing / stripped
                             "key": "file-key-xyz"})
            # env overrides the file, per field.
            os.environ["ECHO_KEY"] = "env-key-123"
            check("env ECHO_KEY overrides config", cfg.resolve_key() == "env-key-123")
            os.environ.pop("ECHO_KEY", None)
            # missing required field raises a helpful error.
            cfg.write_config("Owner Name", "", "")
            try:
                cfg.resolve_key()
                check("missing key raises", False)
            except RuntimeError:
                check("missing key raises", True)
    finally:
        for k, v in saved_env.items():
            if v is None:
                os.environ.pop(k, None)
            else:
                os.environ[k] = v

    # M2 — slug collision + link-confidence helpers are implemented.
    import echo_quality as qual
    check("M2 safe_slug disambiguates a collision",
          qual.safe_slug({"foo"}, "foo") != "foo")
    check("M2 short/common tokens are not confident links",
          qual.is_confident_link("API", "we built an api") is False)

    # M3 — doctor module exposes run(); M4 — output envelope is implemented.
    import echo_doctor as doc
    check("M3 doctor exposes run()", hasattr(doc, "run"))
    import echo_output as out
    env = out.envelope("created", {"path": "p.md"})
    check("M4 envelope shape", env.get("ok") is True and env.get("action") == "created")

    print(f"\n{len(failures)} failure(s)" if failures else "\nall scaffold interface checks passed")
    return 1 if failures else 0


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