#!/usr/bin/env python3 """mock_olrapi.py — a deterministic mock of the Obsidian Local REST API surface the echo-memory plugin uses, with controllable fault injection. It reproduces the real API's observed behavior and quirks so the A/B eval can run without credentials and without touching the live vault: GET /vault/ 200 + body, or 404 {errorCode:40400} GET /vault/ (or dir + '/') 200 {"files":[... , "sub/"]} (dirs end in '/') GET /vault// 400 (the double-slash quirk) GET ... Accept: document-map 200 {"headings":[...]} PUT /vault/ 201/200, stores body, creates parents POST /vault/ 200, appends (creates if absent) PATCH/vault/ heading target missing -> 400 {errorCode:40080}; else apply POST /search/simple/?query= 200 [] DELETE /vault/ 200 / 404 Fault injection is triggered by markers in the path (so a single server serves every scenario deterministically): * path contains "flaky" -> the FIRST write (PUT/POST) to that path returns 503, subsequent writes succeed (tests retry-on-5xx). * path contains "phantom" -> PUT returns 200 but does NOT persist (accepted-but-lost; tests read-back verify). * a PATCH heading whose Target heading is absent from the stored doc -> 400 40080 (the silent-write-loss trigger). Debug (out of band, for the harness ground-truth checks): GET /__debug__?path=

-> raw stored content, or "<>" POST /__debug__reset -> clear all state + fault memory """ import json, re, sys, argparse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs, unquote STATE = {} # vault-path -> content (str) FLAKY_FIRED = set() # paths that already consumed their one-time 503 def headings_of(text): return [m.group(1).strip() for m in re.finditer(r"(?m)^#{1,6}\s+(.*)$", text or "")] def heading_present(text, target): # target may be a full "A::B" path; match its leaf heading line anywhere in the doc leaf = target.split("::")[-1].strip() return any(h == leaf or h.endswith("::" + leaf) or h == target for h in headings_of(text)) \ or bool(re.search(r"(?m)^#{1,6}\s+" + re.escape(leaf) + r"\s*$", text or "")) class H(BaseHTTPRequestHandler): def log_message(self, *a): pass # quiet def _send(self, code, body="", ctype="text/markdown"): b = body.encode() if isinstance(body, str) else body self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) def _json(self, code, obj): self._send(code, json.dumps(obj, indent=2), "application/json") def _read_body(self): n = int(self.headers.get("Content-Length", 0) or 0) return self.rfile.read(n).decode("utf-8", "replace") if n else "" def _vpath(self): p = urlparse(self.path).path if p.startswith("/vault/"): return p[len("/vault/"):] # keep trailing slash if present return None # ---- debug ------------------------------------------------------------- def _maybe_debug(self): u = urlparse(self.path) if u.path == "/__debug__": q = parse_qs(u.query) path = unquote(q.get("path", [""])[0]) self._send(200, STATE.get(path, "<>")) return True if u.path == "/__debug__reset": STATE.clear(); FLAKY_FIRED.clear() self._json(200, {"ok": True}) return True return False def do_GET(self): if self._maybe_debug(): return raw = urlparse(self.path).path if raw.startswith("/search/simple"): return self._json(200, []) vp = self._vpath() if vp is None: return self._json(404, {"errorCode": 40400, "message": "Not Found"}) if "//" in self.path.replace("/vault/", "", 1): # double-slash quirk return self._json(400, {"errorCode": 40000, "message": "Bad Request"}) if vp == "" or vp.endswith("/"): # directory listing prefix = vp kids = set() for k in STATE: if k.startswith(prefix) and k != prefix: rest = k[len(prefix):] kids.add(rest.split("/")[0] + ("/" if "/" in rest else "")) return self._json(200, {"files": sorted(kids)}) if vp in STATE: if "document-map" in (self.headers.get("Accept", "")): return self._json(200, {"headings": headings_of(STATE[vp]), "blocks": [], "frontmatterFields": []}) return self._send(200, STATE[vp]) return self._json(404, {"errorCode": 40400, "message": "Not Found"}) def _flaky_once(self, vp): if "flaky" in vp and vp not in FLAKY_FIRED: FLAKY_FIRED.add(vp) return True return False def do_PUT(self): vp = self._vpath(); body = self._read_body() if vp is None: return self._json(400, {"message": "bad path"}) if self._flaky_once(vp): return self._json(503, {"message": "Service Unavailable"}) if "phantom" in vp: # accepted but NOT persisted return self._send(200, "") STATE[vp] = body return self._send(200, "") def do_POST(self): if self._maybe_debug(): return raw = urlparse(self.path).path if raw.startswith("/search/simple"): return self._json(200, []) vp = self._vpath(); body = self._read_body() if vp is None: return self._json(400, {"message": "bad path"}) if self._flaky_once(vp): return self._json(503, {"message": "Service Unavailable"}) STATE[vp] = STATE.get(vp, "") + body # append return self._send(200, "") def do_PATCH(self): vp = self._vpath(); body = self._read_body() if vp is None: return self._json(400, {"message": "bad path"}) ttype = self.headers.get("Target-Type", "") op = self.headers.get("Operation", "append") target = self.headers.get("Target", "") cur = STATE.get(vp, "") if ttype == "heading": if not heading_present(cur, target): return self._json(400, {"errorCode": 40080, "message": "invalid-target"}) # apply: naive append/prepend/replace around the heading line STATE[vp] = cur + ("\n" + body if op != "prepend" else body + "\n") return self._send(200, "") if ttype == "frontmatter": STATE[vp] = cur + f"\n" return self._send(200, "") STATE[vp] = cur + "\n" + body return self._send(200, "") def do_DELETE(self): vp = self._vpath() if vp in STATE: del STATE[vp]; return self._send(200, "") return self._json(404, {"errorCode": 40400, "message": "Not Found"}) def main(): ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8799) a = ap.parse_args() srv = ThreadingHTTPServer(("127.0.0.1", a.port), H) print(f"mock-olrapi listening on http://127.0.0.1:{a.port}", flush=True) srv.serve_forever() if __name__ == "__main__": main()