diff --git a/taosmd/__init__.py b/taosmd/__init__.py index 7a03a695..5367c142 100644 --- a/taosmd/__init__.py +++ b/taosmd/__init__.py @@ -23,6 +23,11 @@ from .catalog_pipeline import CatalogPipeline from .retrieval import retrieve from .api import ingest, search, list_pending_decisions, resolve_pending_decision + +# Activation surfaces — shared service layer + local HTTP/REST API (#85). +# The MCP server (#84) reuses the same `service` core. +from . import service +from .http_server import serve from .cross_encoder import CrossEncoderReranker from .access_tracker import AccessTracker from .preference_extractor import extract_preferences @@ -86,6 +91,9 @@ def agent_rules() -> str: "retrieve", "ingest", "search", + # Activation surfaces + "service", + "serve", "CrossEncoderReranker", "classify_intent", "get_search_strategy", diff --git a/taosmd/cli.py b/taosmd/cli.py index a77eec41..c4a5a6b4 100644 --- a/taosmd/cli.py +++ b/taosmd/cli.py @@ -449,7 +449,33 @@ def main(argv: list[str] | None = None) -> int: help="Free-form note attached to the resolution", ) + # ----- serve subcommand (local HTTP/REST API) ----------------------- + serve_p = sub.add_parser( + "serve", + help="Run the local HTTP/REST memory API (stdlib server, zero deps)", + ) + serve_p.add_argument( + "--host", default="127.0.0.1", + help="Bind address (default 127.0.0.1, localhost-only). " + "Use 0.0.0.0 to expose on the LAN — no auth, so gate it yourself.", + ) + serve_p.add_argument( + "--port", type=int, default=7833, + help="Bind port (default 7833)", + ) + serve_p.add_argument( + "--serve-data-dir", dest="serve_data_dir", default=None, + help="Data dir for served memory (default: $TAOSMD_DATA_DIR or ~/.taosmd)", + ) + args = parser.parse_args(argv) + + if args.cmd == "serve": + from . import http_server # noqa: PLC0415 + return http_server.serve( + host=args.host, port=args.port, data_dir=args.serve_data_dir, + ) + registry = AgentRegistry(args.data_dir) if args.cmd == "review": diff --git a/taosmd/http_server.py b/taosmd/http_server.py new file mode 100644 index 00000000..17a3cbe4 --- /dev/null +++ b/taosmd/http_server.py @@ -0,0 +1,275 @@ +"""Local HTTP/REST API for taOSmd memory (stdlib only, zero deps). + +An activation surface so non-Python apps and remote-on-LAN agents can read +and write taOSmd memory without embedding the Python package — the spirit of +``qmd serve``. It is a thin JSON shell over :mod:`taosmd.service`, the same +shared core the upcoming MCP server (#84) sits on, so behaviour matches the +Python API and CLI exactly. + +Design choices (matching the project's local-first, offline, additive vision): + +* **stdlib only** — :class:`http.server.ThreadingHTTPServer` + + :class:`~http.server.BaseHTTPRequestHandler`, :mod:`json`, no new deps. +* **local-first** — binds ``127.0.0.1`` by default. Pass a different host + (e.g. ``0.0.0.0``) only to expose it on the LAN. There is **no auth**: on + localhost that is fine (any local process already has the Python API); if + you bind to a routable address, put it behind your own network controls. +* **additive + opt-in** — the server only runs when you start it; the + Python API, CLI, and standalone use are untouched. +* **per-agent scoping** — every endpoint takes an ``agent`` and forwards it + to the service layer, honouring the same isolation as the Python API. + +Concurrency model +----------------- +:class:`ThreadingHTTPServer` hands each request to its own thread, but the +underlying stores hold thread-affine SQLite connections (created and cached +on first use). So rather than ``asyncio.run`` per request — which would bind +those connections to a request thread that later disappears — every async +service call is dispatched onto a single, long-lived background event-loop +thread (see :class:`_ServiceLoop`). All DB work therefore happens in one +context, sequentially, exactly like the single-threaded Python API. The +result is marshalled back to the calling request thread. + +Endpoints +--------- +``GET /health`` -> ``{"status": "ok", "version": }`` +``POST /ingest`` ``{"text", "agent"}`` -> ingest result +``POST /search`` ``{"query", "agent", "limit"?}`` -> ``{"hits": [...]}`` +``GET /search?q=&agent=&limit=`` -> ``{"hits": [...]}`` +``GET /pending?agent=`` -> ``{"pending": [...]}`` +``POST /pending/resolve`` ``{"id", "decision", "note"?}`` -> resolve result +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlsplit + +from . import __version__, service + +logger = logging.getLogger(__name__) + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 7833 + + +class _BadRequest(Exception): + """Raised for malformed input -> 400.""" + + +class _ServiceLoop: + """A single background thread running one asyncio event loop. + + All store/DB work runs here so the thread-affine SQLite connections are + created and used in exactly one thread, no matter which request thread + issued the call. ``run(coro)`` blocks the caller until the coroutine + completes and returns its result (or re-raises its exception). + """ + + def __init__(self) -> None: + self._loop = asyncio.new_event_loop() + self._thread = threading.Thread( + target=self._loop.run_forever, name="taosmd-service-loop", daemon=True, + ) + self._thread.start() + + def run(self, coro): + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + return future.result() + + def close(self) -> None: + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(timeout=5) + self._loop.close() + + +def _make_handler(data_dir, runner: _ServiceLoop): + """Build a handler class bound to a fixed ``data_dir``. + + ThreadingHTTPServer instantiates the handler per request, so the data dir + is closed over here rather than threaded through every call site. + """ + + class TaosmdHandler(BaseHTTPRequestHandler): + server_version = f"taosmd/{__version__}" + + # ----- plumbing ---------------------------------------------------- + def log_message(self, fmt, *args): # noqa: A002 - stdlib signature + logger.info("%s - %s", self.address_string(), fmt % args) + + def _send_json(self, status: int, payload: dict) -> None: + body = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if self.command != "HEAD": + self.wfile.write(body) + + def _read_json_body(self) -> dict: + length = int(self.headers.get("Content-Length") or 0) + if length <= 0: + return {} + raw = self.rfile.read(length) + try: + parsed = json.loads(raw.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise _BadRequest(f"invalid JSON body: {exc}") from exc + if not isinstance(parsed, dict): + raise _BadRequest("JSON body must be an object") + return parsed + + # ----- routing ----------------------------------------------------- + def do_GET(self) -> None: # noqa: N802 - stdlib signature + self._dispatch("GET") + + def do_HEAD(self) -> None: # noqa: N802 + self._dispatch("GET") + + def do_POST(self) -> None: # noqa: N802 + self._dispatch("POST") + + def _dispatch(self, method: str) -> None: + parts = urlsplit(self.path) + path = parts.path.rstrip("/") or "/" + query = parse_qs(parts.query) + try: + if method == "GET" and path == "/health": + self._send_json(200, {"status": "ok", "version": __version__}) + elif method == "GET" and path == "/search": + self._handle_search_get(query) + elif method == "POST" and path == "/search": + self._handle_search_post() + elif method == "POST" and path == "/ingest": + self._handle_ingest() + elif method == "GET" and path == "/pending": + self._handle_pending(query) + elif method == "POST" and path == "/pending/resolve": + self._handle_pending_resolve() + else: + self._send_json(404, {"error": f"unknown route: {method} {path}"}) + except _BadRequest as exc: + self._send_json(400, {"error": str(exc)}) + except ValueError as exc: + # Service-layer validation (e.g. missing agent, bad action). + self._send_json(400, {"error": str(exc)}) + except Exception as exc: # noqa: BLE001 - surface as 500 JSON + logger.exception("taosmd http: unhandled error for %s %s", method, path) + self._send_json(500, {"error": f"{type(exc).__name__}: {exc}"}) + + # ----- handlers ---------------------------------------------------- + def _handle_ingest(self) -> None: + body = self._read_json_body() + text = body.get("text") + agent = body.get("agent") + if not isinstance(text, str) or not text: + raise _BadRequest("'text' (non-empty string) is required") + if not isinstance(agent, str) or not agent: + raise _BadRequest("'agent' (non-empty string) is required") + result = runner.run(service.ingest(text, agent=agent, data_dir=data_dir)) + self._send_json(200, result) + + def _handle_search_post(self) -> None: + body = self._read_json_body() + query = body.get("query") + agent = body.get("agent") + limit = body.get("limit", 5) + self._do_search(query, agent, limit) + + def _handle_search_get(self, qs: dict) -> None: + query = (qs.get("q") or qs.get("query") or [None])[0] + agent = (qs.get("agent") or [None])[0] + limit = (qs.get("limit") or [5])[0] + self._do_search(query, agent, limit) + + def _do_search(self, query, agent, limit) -> None: + if not isinstance(query, str) or not query: + raise _BadRequest("'query' (non-empty string) is required") + if not isinstance(agent, str) or not agent: + raise _BadRequest("'agent' (non-empty string) is required") + try: + limit_i = int(limit) + except (TypeError, ValueError) as exc: + raise _BadRequest("'limit' must be an integer") from exc + hits = runner.run( + service.search(query, agent=agent, data_dir=data_dir, limit=limit_i) + ) + self._send_json(200, {"hits": hits}) + + def _handle_pending(self, qs: dict) -> None: + agent = (qs.get("agent") or [None])[0] + limit = (qs.get("limit") or [20])[0] + try: + limit_i = int(limit) + except (TypeError, ValueError) as exc: + raise _BadRequest("'limit' must be an integer") from exc + pending = runner.run( + service.pending_list(agent=agent, data_dir=data_dir, limit=limit_i) + ) + self._send_json(200, {"pending": pending}) + + def _handle_pending_resolve(self) -> None: + body = self._read_json_body() + decision_id = body.get("id") + decision = body.get("decision") + note = body.get("note", "") + if not isinstance(decision_id, str) or not decision_id: + raise _BadRequest("'id' (non-empty string) is required") + if decision not in {"accept", "reject", "modify"}: + raise _BadRequest("'decision' must be one of accept|reject|modify") + result = runner.run( + service.pending_resolve( + decision_id, decision, note=note or "", data_dir=data_dir, + ) + ) + self._send_json(200, result) + + return TaosmdHandler + + +def make_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=None): + """Create (but do not start) a :class:`ThreadingHTTPServer`. + + Useful for tests that need to bind an ephemeral port (``port=0``) and read + back the assigned address before serving. Call ``server.serve_forever()`` + to run it, ``server.shutdown()`` + ``server.server_close()`` to stop. + + A :class:`_ServiceLoop` is started and attached as ``server.service_loop``; + closing it is handled by :func:`serve`. Tests that drive ``make_server`` + directly should call ``server.service_loop.close()`` during teardown. + """ + runner = _ServiceLoop() + httpd = ThreadingHTTPServer((host, port), _make_handler(data_dir, runner)) + httpd.service_loop = runner + return httpd + + +def serve(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, data_dir=None) -> int: + """Run the local HTTP memory server until interrupted. + + Binds ``host:port`` (default ``127.0.0.1:7833``) and blocks serving + requests. Returns 0 on a clean Ctrl-C shutdown. + """ + httpd = make_server(host, port, data_dir) + bound_host, bound_port = httpd.server_address[:2] + where = "localhost only" if bound_host in {"127.0.0.1", "::1"} else "LAN-reachable (no auth)" + print(f"taosmd HTTP API listening on http://{bound_host}:{bound_port} ({where})") + print("Endpoints: GET /health, POST /ingest, GET|POST /search, " + "GET /pending, POST /pending/resolve") + try: + httpd.serve_forever() + except KeyboardInterrupt: + print("\nshutting down…") + finally: + httpd.shutdown() + httpd.server_close() + httpd.service_loop.close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(serve()) diff --git a/taosmd/service.py b/taosmd/service.py new file mode 100644 index 00000000..2bc9c690 --- /dev/null +++ b/taosmd/service.py @@ -0,0 +1,109 @@ +"""Adapter-agnostic service layer over :mod:`taosmd.api`. + +This is the shared core that activation surfaces sit on top of: the local +HTTP/REST server (#85) and the upcoming MCP server (#84) both call these +functions rather than reaching into :mod:`taosmd.api` directly. Keeping the +glue in one place means a single, consistent contract for every adapter and +guarantees their behaviour matches the Python API exactly. + +The functions here are deliberately thin — they reuse +:func:`taosmd.api.ingest`, :func:`taosmd.api.search`, +:func:`taosmd.api.list_pending_decisions`, and +:func:`taosmd.api.resolve_pending_decision` (and therefore +``_ensure_stores`` / the stores cache / ``TAOSMD_DATA_DIR`` handling) so +the only thing they add is a uniform, transport-friendly signature: +``(positional, agent=..., data_dir=..., **opts)``. +""" + +from __future__ import annotations + +from . import api as _api + + +async def ingest(text, *, agent: str, data_dir=None, **opts) -> dict: + """Shelve a transcript and embed it for later search. + + Thin wrapper over :func:`taosmd.api.ingest`. ``text`` may be a string, + a turn dict, or an iterable of either (see the underlying API for the + accepted shapes). Returns ``{"archived", "agent", "data_dir"}``. + """ + return await _api.ingest(text, agent=agent, data_dir=data_dir, **opts) + + +async def search(query: str, *, agent: str, data_dir=None, limit: int = 5, **opts) -> list[dict]: + """Search memory for passages relevant to ``query``. + + Thin wrapper over :func:`taosmd.api.search`. Returns ranked hits in the + agent-rules contract shape (``text``/``source``/``timestamp``/ + ``confidence``/``metadata``). + """ + return await _api.search(query, agent=agent, data_dir=data_dir, limit=limit, **opts) + + +async def pending_list(*, agent: str | None = None, data_dir=None, limit: int = 20) -> list[dict]: + """Return unresolved KG-update decisions deferred by the librarian. + + ``agent`` is accepted for adapter symmetry; the pending-decisions queue + is keyed per data dir (per install), not per agent, so it is not used to + filter here. Use ``subject=`` on the underlying API if subject-level + filtering is needed. + """ + return await _api.list_pending_decisions(limit=limit, data_dir=data_dir) + + +async def pending_resolve( + decision_id: str, + decision: str, + *, + note: str = "", + data_dir=None, +) -> dict: + """Resolve a pending decision with the user's explicit choice. + + ``decision`` is one of ``accept`` / ``reject`` / ``modify`` (the + ``action`` argument of :func:`taosmd.api.resolve_pending_decision`). + Returns ``{ok, applied_kg, resolution}``. + """ + return await _api.resolve_pending_decision( + decision_id, action=decision, note=note, data_dir=data_dir, + ) + + +async def stats(*, agent: str, data_dir=None) -> dict: + """Return lightweight stats for an agent. + + Ensures the stores exist (so a freshly-pointed data dir is initialised), + then reports the registry record for ``agent`` plus the resolved data + dir. Shape: ``{"agent", "data_dir", "registered", "created_at", + "last_ingest_at", "total_chunks"}``. Unknown agents report + ``registered=False`` with zeroed counters rather than raising, so the + surface stays forgiving for read-only probes. + """ + if not agent: + raise ValueError("agent name is required") + stores = await _api._ensure_stores(data_dir) + + from .agents import AgentNotFoundError, get_agent # noqa: PLC0415 + + out = { + "agent": agent, + "data_dir": stores["data_dir"], + "registered": False, + "created_at": 0, + "last_ingest_at": 0, + "total_chunks": 0, + } + try: + record = get_agent(agent) + except AgentNotFoundError: + return out + out.update( + registered=True, + created_at=record.get("created_at", 0), + last_ingest_at=record.get("last_ingest_at", 0), + total_chunks=record.get("total_chunks", 0), + ) + return out + + +__all__ = ["ingest", "search", "pending_list", "pending_resolve", "stats"] diff --git a/tests/test_http_server.py b/tests/test_http_server.py new file mode 100644 index 00000000..4f46e5fa --- /dev/null +++ b/tests/test_http_server.py @@ -0,0 +1,158 @@ +"""Tests for taosmd.http_server — the local HTTP/REST activation surface. + +Offline + fast: the server runs in a background thread on an ephemeral port +with an isolated tmp data dir, and the vector embedder is patched (same +deterministic hash vector as tests/test_api.py) so no ONNX/QMD model is +needed. Requests go over the loopback via urllib. +""" + +from __future__ import annotations + +import json +import threading +import urllib.error +import urllib.request +from pathlib import Path + +import pytest + +from taosmd import api as taosmd_api +from taosmd import http_server + + +def _patch_embedder(stores: dict) -> None: + """Deterministic 8-dim hash embedder so search finds matching text.""" + vmem = stores["vector"] + + async def _fake_embed(text: str, task: str = "search_document") -> list[float]: + h = hash(text) & 0xFFFFFFFF + return [((h >> (i * 4)) & 0xFF) / 255.0 for i in range(8)] + + vmem.embed = _fake_embed # type: ignore[assignment] + + +@pytest.fixture +def live_server(tmp_path, monkeypatch): + """Start the HTTP server on an ephemeral port against an isolated data dir. + + Yields the base URL (e.g. ``http://127.0.0.1:54321``). Tears the server + and the cached SQLite stores down cleanly afterwards. + """ + data_dir = tmp_path / "taosmd-data" + data_dir.mkdir() + monkeypatch.setattr(taosmd_api, "_stores_cache", {}) + + httpd = http_server.make_server("127.0.0.1", 0, data_dir=str(data_dir)) + + # Init the stores *on the server's service loop thread* (so the thread- + # affine SQLite connections live where the handlers will use them), then + # patch the embedder so search doesn't need a real ONNX/QMD model. + stores = httpd.service_loop.run(taosmd_api._ensure_stores(str(data_dir))) + _patch_embedder(stores) + + host, port = httpd.server_address[:2] + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + try: + yield f"http://{host}:{port}" + finally: + httpd.shutdown() + httpd.server_close() + thread.join(timeout=5) + for s in list(taosmd_api._stores_cache.values()): + for store in (s.get("archive"), s.get("vector"), s.get("kg")): + if store and hasattr(store, "close"): + try: + httpd.service_loop.run(store.close()) + except Exception: + pass + httpd.service_loop.close() + + +def _post(url: str, payload) -> tuple[int, dict]: + data = payload if isinstance(payload, (bytes, bytearray)) else json.dumps(payload).encode() + req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST") + return _send(req) + + +def _get(url: str) -> tuple[int, dict]: + return _send(urllib.request.Request(url, method="GET")) + + +def _send(req) -> tuple[int, dict]: + try: + with urllib.request.urlopen(req, timeout=10) as resp: + return resp.status, json.loads(resp.read().decode()) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode()) + + +def test_health(live_server): + status, body = _get(f"{live_server}/health") + assert status == 200 + assert body["status"] == "ok" + assert isinstance(body["version"], str) and body["version"] + + +def test_ingest_then_search_roundtrip(live_server): + status, body = _post( + f"{live_server}/ingest", + {"text": "The HTTP API ships on the feat/http-api branch.", "agent": "http-test"}, + ) + assert status == 200, body + assert body["archived"] == 1 + assert body["agent"] == "http-test" + + status, body = _post( + f"{live_server}/search", + {"query": "The HTTP API ships on the feat/http-api branch.", "agent": "http-test"}, + ) + assert status == 200, body + assert body["hits"], "expected the ingested text to be retrievable" + assert "HTTP API" in body["hits"][0]["text"] + + +def test_search_get_query_param(live_server): + _post( + f"{live_server}/ingest", + {"text": "GET-style search works over query params.", "agent": "http-test"}, + ) + status, body = _get( + f"{live_server}/search?q=GET-style%20search%20works%20over%20query%20params.&agent=http-test&limit=3" + ) + assert status == 200, body + assert body["hits"] + assert "GET-style" in body["hits"][0]["text"] + + +def test_bad_json_body_returns_400(live_server): + status, body = _post(f"{live_server}/ingest", b"{not valid json") + assert status == 400 + assert "error" in body + + +def test_missing_field_returns_400(live_server): + status, body = _post(f"{live_server}/ingest", {"text": "no agent here"}) + assert status == 400 + assert "agent" in body["error"] + + +def test_unknown_route_returns_404(live_server): + status, body = _get(f"{live_server}/does-not-exist") + assert status == 404 + assert "error" in body + + +def test_pending_empty(live_server): + status, body = _get(f"{live_server}/pending?agent=http-test") + assert status == 200 + assert body["pending"] == [] + + +def test_pending_resolve_bad_decision_returns_400(live_server): + status, body = _post( + f"{live_server}/pending/resolve", + {"id": "abc", "decision": "frobnicate"}, + ) + assert status == 400 + assert "decision" in body["error"]