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")))

    # M1 — secret resolution order is implemented; env wins, then the credentials file.
    import echo_secrets as s
    os.environ["ECHO_KEY"] = "env-key-123"
    check("M1 env ECHO_KEY wins", s.resolve_key("legacy-fallback") == "env-key-123")
    with tempfile.TemporaryDirectory() as d:
        os.environ["ECHO_STATE_DIR"] = d
        os.environ.pop("ECHO_KEY", None)
        s.write_key("file-key-xyz")
        check("M1 write_key -> credentials -> resolve_key reads it", s.resolve_key() == "file-key-xyz")
        os.environ["ECHO_KEY_LEGACY_OK"] = "1"  # silence the warning for the legacy-path check
        os.environ["ECHO_STATE_DIR"] = os.path.join(d, "empty")  # an empty dir: no creds file
        check("M1 legacy fallback when env+file absent", s.resolve_key("legacy-fallback") == "legacy-fallback")
    os.environ["ECHO_KEY"] = "env-key-123"  # restore for any later checks

    # 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())