#!/usr/bin/env python3 """test_ops_api.py โ€” Phase 0 (return-not-print) contract: every high-level op has a core `*_op` function that RETURNS an envelope dict and prints nothing to stdout. This is the seam the MCP server wraps (docs/MCP-SERVER-SPEC.md ยง4): the CLI wrappers are tested by the other suites; here we call the cores in-process against the mock and assert (a) the envelope shapes and (b) stdout purity โ€” a stray print() would corrupt an MCP stdio/HTTP response stream. Run: python test_ops_api.py [--port 8850] """ from __future__ import annotations import argparse import contextlib import io import json import os import subprocess import sys import tempfile import time import urllib.request from pathlib import Path HERE = Path(__file__).resolve().parent SCRIPTS = HERE.parent / "echo-memory.plugin.src" / "skills" / "echo-memory" / "scripts" KEY = "test-key-not-a-real-secret" ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8850) a = ap.parse_args() BASE = f"http://127.0.0.1:{a.port}" # env must be set BEFORE importing echo (it resolves config at import time) os.environ.update(ECHO_BASE=BASE, ECHO_KEY=KEY, ECHO_TODAY="2026-06-21", ECHO_NOW="2300", ECHO_STATE_DIR=tempfile.mkdtemp(), ECHO_VERIFY="0") sys.path.insert(0, str(SCRIPTS)) failures = [] def check(name, cond, detail=""): print(f"{'ok ' if cond else 'FAIL'} {name}" + (f" -- {detail}" if not cond else "")) if not cond: failures.append(name) def http(method, url, body=None): data = body.encode() if isinstance(body, str) else body req = urllib.request.Request(url, data=data, method=method, headers={"Authorization": f"Bearer {KEY}"}) try: with urllib.request.urlopen(req, timeout=10) as r: return r.status, r.read().decode("utf-8", "replace") except Exception as e: # noqa: BLE001 return getattr(e, "code", 0), "" def pure(fn, *args, **kw): """Call fn capturing stdout; return (result, captured_stdout).""" buf = io.StringIO() with contextlib.redirect_stdout(buf): out = fn(*args, **kw) return out, buf.getvalue() def main(): srv = subprocess.Popen([sys.executable, str(HERE / "mock_olrapi.py"), "--port", str(a.port)], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) try: for _ in range(50): try: urllib.request.urlopen(f"{BASE}/__debug__reset", data=b"", timeout=1) break except Exception: time.sleep(0.1) http("PUT", f"{BASE}/vault/_agent/echo-vault.md", "---\nschema_version: 4\n---\n# marker\n") http("PUT", f"{BASE}/vault/_agent/context/current-context.md", "---\ntype: context-bundle\nscope_updated: \"2026-06-20\"\ncreated: 2026-06-01\n---\n" "# Current Context\n\n## Scope\ntesting phase 0\n\n## Scope History\n") import echo import echo_doctor import echo_ops import echo_recall import echo_reflect import echo_session import echo_triage # capture_op: create / duplicate-gate / dry-run โ€” envelopes, stdout pure env, out = pure(echo_ops.capture_op, "person", "Zara Quix", body_text="Met at the expo.") check("capture_op create envelope", env.get("ok") is True and env.get("action") == "created" and env.get("path") == "resources/people/zara-quix.md", json.dumps(env)) check("capture_op prints nothing to stdout", out == "", out[:200]) env, out = pure(echo_ops.capture_op, "person", "Zara Quix Junior", body_text="") check("capture_op duplicate-gate is data, not an exit code", env.get("ok") is False and env.get("action") == "duplicate-gate" and env.get("candidates"), json.dumps(env)) check("capture_op gate prints nothing", out == "", out[:200]) env, out = pure(echo_ops.capture_op, "company", "Plan Co", body_text="", dry_run=True) check("capture_op dry-run envelope", env.get("action") == "dry-run:create" and env.get("path") == "resources/companies/plan-co.md", json.dumps(env)) # resolve_op / link_op env, out = pure(echo_ops.resolve_op, "zara quix") check("resolve_op returns the match dict", env.get("match") is True and env.get("path") == "resources/people/zara-quix.md", json.dumps(env)) pure(echo_ops.capture_op, "concept", "Gizmo", body_text="") env, out = pure(echo_ops.link_op, "resources/people/zara-quix.md", "resources/concepts/gizmo.md") check("link_op envelope", env.get("ok") is True and env.get("a_changed") in (True, False), json.dumps(env)) check("link_op prints nothing", out == "", out[:200]) # recall_op env, out = pure(echo_recall.recall_op, "expo") check("recall_op envelope shape", env.get("action") == "recall" and isinstance(env.get("primary"), list) and isinstance(env.get("linked"), list), json.dumps(env)[:200]) check("recall_op prints nothing", out == "", out[:200]) # triage list_op / route_op (dry-run) http("PUT", f"{BASE}/vault/inbox/captures/inbox.md", "- 2026-06-10: try the quorlab tool\n") env, out = pure(echo_triage.list_op) check("triage list_op envelope", env.get("count") == 1 and env["items"][0]["age_days"] == 11, json.dumps(env)) props = [{"title": "Quorlab Tool", "kind": "reference", "confidence": 0.9, "line": "- 2026-06-10: try the quorlab tool"}] env, out = pure(echo_triage.route_op, props, False) check("triage route_op dry-run rows", env.get("dry_run") is True and env["rows"][0]["action"] == "create", json.dumps(env)) check("triage route_op prints nothing", out == "", out[:200]) # reflect apply_op (applied path, using capture_op internally) env, out = pure(echo_reflect.apply_op, [{"title": "Blorp Pattern", "kind": "semantic", "body": "The operator prefers blorp.", "confidence": 0.9}], True) check("reflect apply_op applies via capture_op", env.get("applied") == 1 and env["results"][0]["action"] == "created", json.dumps(env)) check("reflect apply_op prints nothing", out == "", out[:200]) # scope ops env, out = pure(echo.scope_show_op) check("scope_show_op returns scope + freshness", env.get("scope") == "testing phase 0" and env.get("scope_updated") == "2026-06-20", json.dumps(env)) env, out = pure(echo.scope_set_op, "phase zero refactor") check("scope_set_op envelope", env.get("action") == "scope-set" and env.get("scope_updated") == "2026-06-21", json.dumps(env)) check("scope_set_op prints nothing", out == "", out[:200]) # load_op (brief) env, out = pure(echo.load_op, True) check("load_op returns sections + brief", "marker" in env.get("sections", {}) and "ECHO load (brief)" in env.get("brief", ""), json.dumps(env)[:200]) check("load_op prints nothing", out == "", out[:200]) # doctor run_op env, out = pure(echo_doctor.run_op) check("doctor run_op checks list", isinstance(env.get("checks"), list) and env.get("fatal") is None and env.get("ok") is True, json.dumps(env)) check("doctor run_op prints nothing", out == "", out[:200]) # session_end_op: dry-run envelope; bad bundle raises EchoError(2) pre-write bundle = {"slug": "phase-zero", "log_body": "---\ntype: session-log\n---\n# S\n\n## Goal\nx\n"} env, out = pure(echo_session.session_end_op, bundle, False) check("session_end_op dry-run envelope", env.get("dry_run") is True and env.get("path") == "_agent/sessions/2026-06-21-2300-phase-zero.md", json.dumps(env)) check("session_end_op prints nothing", out == "", out[:200]) try: echo_session.session_end_op({"slug": "Bad Slug!", "log_body": "x"}, True) check("session_end_op rejects a bad slug", False, "no exception raised") except RuntimeError as exc: check("session_end_op rejects a bad slug", getattr(exc, "code", 0) == 2, str(exc)) print(f"\n{len(failures)} failure(s)" if failures else "\nall ops-api tests passed") return 1 if failures else 0 finally: srv.terminate() if __name__ == "__main__": raise SystemExit(main())