diff --git a/README.md b/README.md index 62fb52c..f2a1b47 100644 --- a/README.md +++ b/README.md @@ -133,23 +133,15 @@ Use the absolute path. `~` and `$HOME` won't expand here. For Claude Desktop (untested), edit `~/Library/Application Support/Claude/claude_desktop_config.json`. -### Install the OpenCode plugins +### Install the OpenCode plugin -**Capture plugin** — listens for idle sessions and writes a deferred-capture JSONL file for the daemon to drain. +This is the OpenCode equivalent of the Claude Code Stop hook. It listens for idle sessions and writes a deferred-capture JSONL file for the daemon to drain. ```bash mkdir -p ~/.config/opencode/plugins cp deploy/opencode/iai-mcp-capture.js ~/.config/opencode/plugins/ ``` -**Memory-inject plugin** — automatically loads session background memory on every new session via the daemon's HTTP adapter. Memory is injected into the model's system prompt (clean, no phantom messages, preserves session title). - -```bash -cp deploy/opencode/iai-mcp-memory-inject.js ~/.config/opencode/plugins/ -``` - -This plugin requires the daemon's HTTP listener. Set `IAI_DAEMON_HTTP_PORT=0` in your daemon's systemd/launchd config and restart it. The daemon writes the live port to `~/.iai-mcp/.http.port`, which the plugin reads automatically. - Make sure `@opencode-ai/plugin` is installed: ```bash @@ -157,27 +149,12 @@ cd ~/.config/opencode npm install @opencode-ai/plugin ``` -Register both plugins in `~/.config/opencode/config.json`: - -```json -{ - "plugin": [ - "~/.config/opencode/plugins/iai-mcp-capture.js", - "~/.config/opencode/plugins/iai-mcp-memory-inject.js" - ] -} -``` - ### Connect OpenCode -Add the MCP server and plugins to `~/.config/opencode/config.json`: +Add the MCP server to `~/.config/opencode/config.json`: ```json { - "plugin": [ - "~/.config/opencode/plugins/iai-mcp-capture.js", - "~/.config/opencode/plugins/iai-mcp-memory-inject.js" - ], "mcp": { "iai-mcp": { "type": "local", @@ -188,7 +165,7 @@ Add the MCP server and plugins to `~/.config/opencode/config.json`: } ``` -Use the absolute path to `mcp-wrapper/dist/index.js`. Both plugin files must already be in `~/.config/opencode/plugins/`. +Use the absolute path to `mcp-wrapper/dist/index.js`. The plugin file (`iai-mcp-capture.js`) must already be in `~/.config/opencode/plugins/`. ### Verify @@ -385,7 +362,7 @@ Limitations worth knowing about: Claude Code is the primary host, validated in daily use. -OpenCode is supported via the `iai-mcp-capture.js` and `iai-mcp-memory-inject.js` plugins (see Install the OpenCode plugins above) and MCP server config in `~/.config/opencode/config.json`. The memory-inject plugin requires the daemon's HTTP listener (`IAI_DAEMON_HTTP_PORT` env var). +OpenCode is supported via the `iai-mcp-capture.js` plugin (see Install the OpenCode plugin above) and MCP server config in `~/.config/opencode/config.json`. Claude Desktop should work (uses `claude_desktop_config.json` instead of `~/.claude.json`) but hasn't been tested end to end. diff --git a/deploy/opencode/iai-mcp-memory-inject.js b/deploy/opencode/iai-mcp-memory-inject.js deleted file mode 100644 index 81162e4..0000000 --- a/deploy/opencode/iai-mcp-memory-inject.js +++ /dev/null @@ -1,147 +0,0 @@ -/** - * iai-mcp memory-injection plugin for OpenCode (approach A). - * - * Pulls session-start memory from the iai-mcp daemon's localhost HTTP adapter - * and injects it directly into the model's system prompt via the - * `experimental.chat.system.transform` hook. The memory CONTENT is placed in - * context — no tool call, no injected "INIT" user turn, so the session title - * is generated from the user's real first message (clean). - * - * REPLACES iai-mcp-session-init.js — do NOT run both. session-init forces a - * tool call via a phantom turn (hijacks the title); this plugin needs neither. - * - * Requires the daemon's HTTP listener: - * - set IAI_DAEMON_HTTP_PORT in the daemon's systemd unit (e.g. "0" for an - * OS-assigned port) and restart it. The daemon writes the live port to - * ~/.iai-mcp/.http.port, which this plugin reads. - * - * wake_depth=standard is requested so l0/l1/l2/rich_club carry real content - * (minimal mode returns only opaque handles — nothing worth injecting). - * Override via IAI_MCP_WAKE_DEPTH (minimal|standard|deep). - * - * Fail-safe: any error is swallowed; system-prompt assembly must never break. - */ - -const HOME = process.env.HOME || process.cwd(); -const PORT_FILE = `${HOME}/.iai-mcp/.http.port`; -const WAKE_DEPTH = process.env.IAI_MCP_WAKE_DEPTH || "standard"; -const FETCH_TIMEOUT_MS = 10000; // a cold runtime-graph build can exceed 5s -const MAX_ATTEMPTS = 3; // give up after this many consecutive failed fetches per session -const INJECT_MAX_TURNS = 3; // stop injecting after N turns (token-cost control) - -const memo = new Map(); // sessionID -> injected text | null (null = permanent failure) -const inflight = new Map(); // sessionID -> Promise (await for concurrent callers) -const attempts = new Map(); // sessionID -> consecutive-failed-fetch count -const turns = new Map(); // sessionID -> number of times memory was injected - -async function readPort() { - const fs = await import("node:fs"); - try { - const port = parseInt(fs.readFileSync(PORT_FILE, "utf8").trim(), 10); - return Number.isInteger(port) && port > 0 ? port : null; - } catch { - return null; // daemon HTTP not enabled / not up yet - } -} - -async function fetchMemory(sessionId) { - const port = await readPort(); - if (!port) return { ok: false, text: "" }; - const url = - `http://127.0.0.1:${port}/memory/session-context` + - `?session_id=${encodeURIComponent(sessionId)}` + - `&wake_depth=${encodeURIComponent(WAKE_DEPTH)}&format=text`; - try { - const res = await fetch(url, { signal: AbortSignal.timeout(FETCH_TIMEOUT_MS) }); - const text = (await res.text()).trim(); - // Distinguish network failure from valid-but-empty response. - // An empty body on HTTP 200 is a valid result (new user, minimal mode). - return { ok: res.ok, text }; - } catch { - return { ok: false, text: "" }; - } -} - -// Centralised fetch+cache so the warm-on-create event and the per-turn -// transform hook share ONE in-flight fetch, one cache, and one attempt budget. -// This dedupes concurrent calls so a session never emits duplicate daemon-side -// session_started events. -// -// Fix: inflight stores a Promise so concurrent callers await the same fetch -// instead of getting "". Fix: memo stores null for permanent failure so we -// don't retry forever. Fix: distinguish {ok:true,text:""} (valid empty) from -// {ok:false} (network error). -async function ensureMemory(sessionId) { - // Already resolved — return cached value (may be "" or null). - const cached = memo.get(sessionId); - if (cached !== undefined) return cached; - - // A fetch is already running — await it instead of returning "". - const existing = inflight.get(sessionId); - if (existing) return existing; - - // Permanent failure — stop retrying. - if ((attempts.get(sessionId) || 0) >= MAX_ATTEMPTS) { - memo.set(sessionId, null); - return null; - } - - const promise = (async () => { - const { ok, text } = await fetchMemory(sessionId); - if (ok) { - // HTTP 200 — memoize regardless of content (empty is valid). - memo.set(sessionId, text); - attempts.delete(sessionId); // reset failure counter on success - } else { - // Network error — increment counter for retry budget. - attempts.set(sessionId, (attempts.get(sessionId) || 0) + 1); - } - return text; - })(); - inflight.set(sessionId, promise); - try { - return await promise; - } finally { - inflight.delete(sessionId); - } -} - -export const IaiMcpMemoryInject = async () => { - return { - // Warm-on-create: prime the daemon graph cache (and our memo) as soon as a - // session appears — before the user's first turn — so the first transform - // injects immediately instead of paying the cold-build latency. - event: async ({ event }) => { - if (event.type !== "session.updated") return; - const sid = event.properties?.info?.id; - if (!sid) return; - try { - await ensureMemory(sid); - } catch { - // never throw from an event handler - } - }, - - "experimental.chat.system.transform": async (input, output) => { - try { - const sid = input?.sessionID; - if (!sid || !output || !Array.isArray(output.system)) return; - - // Turn-limited injection: stop after INJECT_MAX_TURNS to control token cost. - // Memory was already in the system prompt for the first N turns; the model - // has seen it and doesn't need repetition every turn. - const currentTurns = turns.get(sid) || 0; - if (currentTurns >= INJECT_MAX_TURNS) return; - - const text = await ensureMemory(sid); - if (text) { - turns.set(sid, currentTurns + 1); - output.system.push(`# iai-mcp memory (session start)\n${text}`); - } - } catch (err) { - // NEVER throw — a plugin error must not break system-prompt assembly. - console.error(`[iai-mcp] memory inject failed: ${err.message}`); - } - }, - }; -}; diff --git a/deploy/opencode/iai-mcp-session-init.js b/deploy/opencode/iai-mcp-session-init.js deleted file mode 100644 index 0949d76..0000000 --- a/deploy/opencode/iai-mcp-session-init.js +++ /dev/null @@ -1,47 +0,0 @@ -/** - * @deprecated Use `iai-mcp-memory-inject.js` instead. - * - * This plugin forces a tool call via a phantom "INIT" user turn, which - * corrupts session title generation. The memory-inject plugin uses the - * system.transform hook to inject memory directly into the system prompt - * — no phantom turn, clean titles, memory visible every turn. - * - * Retained temporarily for backward compatibility. Will be removed in a - * future release. Migrate to iai-mcp-memory-inject.js: - * - * cp deploy/opencode/iai-mcp-memory-inject.js ~/.config/opencode/plugins/ - * - * Then replace this plugin reference in ~/.config/opencode/config.json. - */ - -export const IaiMcpSessionInit = async ({ - client, - project, - directory, - worktree, - experimental_workspace, - serverUrl, - $, -}) => { - const doneSessions = new Set(); - - return { - event: async ({ event }) => { - if (event.type !== "session.updated") return; - const sid = event.properties?.info?.id; - if (!sid || doneSessions.has(sid)) return; - doneSessions.add(sid); - - try { - await client.session.prompt({ - path: { id: sid }, - body: { - system: "MANDATORY: Call `memory_session_context` (no arguments) BEFORE answering any user query.", - parts: [{ type: "text", text: "INIT" }], - noReply: false, - }, - }); - } catch {} - }, - }; -}; diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index f179f3c..4213bc5 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -8,15 +8,16 @@ "name": "iai-mcp-wrapper", "version": "0.1.0", "dependencies": { - "@modelcontextprotocol/sdk": "^1.0.0" + "@modelcontextprotocol/sdk": "^1.0.0", + "zod": "^4.0.0" }, "bin": { "iai-mcp-wrapper": "dist/index.js" }, "devDependencies": { - "@types/node": "^25.0.0", + "@types/node": "^22.0.0", "tsx": "^4.7.0", - "typescript": "^6.0.0" + "typescript": "^5.4.0" }, "engines": { "node": ">=18" @@ -517,13 +518,13 @@ } }, "node_modules/@types/node": { - "version": "25.9.5", - "resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.5.tgz", - "integrity": "sha512-OScDchr2fwuUmWdf4kZ9h7PcJiYDVInhJizG/biAq3cAvqwYktuy/TYGGdZNMtNTFUP7rnb0NU4TUdm82kt4Rg==", + "version": "22.19.17", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.19.17.tgz", + "integrity": "sha512-wGdMcf+vPYM6jikpS/qhg6WiqSV/OhG+jeeHT/KlVqxYfD40iYJf9/AE1uQxVWFvU7MipKRkRv8NSHiCGgPr8Q==", "dev": true, "license": "MIT", "dependencies": { - "undici-types": ">=7.24.0 <7.24.7" + "undici-types": "~6.21.0" } }, "node_modules/accepts": { @@ -1584,9 +1585,9 @@ } }, "node_modules/tsx": { - "version": "4.23.0", - "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.0.tgz", - "integrity": "sha512-eUdUIaCr963q2h5u3+QwvYp0+eqPvn+egeqZUm0hwERCqqx1E3kK5ehbGCvqSE5MQAULr67ww0cA3jKc3YkM1w==", + "version": "4.22.3", + "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.22.3.tgz", + "integrity": "sha512-mdoNxBC/cSQObGGVQ5Bpn5i+yv7j68gk3Nfm3wFjcJg3Z0Mix9jzAFfP12prmm5eVGmDKtp0yyArrs0Q+8gZHg==", "dev": true, "license": "MIT", "dependencies": { @@ -1617,9 +1618,9 @@ } }, "node_modules/typescript": { - "version": "6.0.3", - "resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz", - "integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==", + "version": "5.9.3", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", + "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", "dev": true, "license": "Apache-2.0", "bin": { @@ -1631,9 +1632,9 @@ } }, "node_modules/undici-types": { - "version": "7.24.6", - "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.24.6.tgz", - "integrity": "sha512-WRNW+sJgj5OBN4/0JpHFqtqzhpbnV0GuB+OozA9gCL7a993SmU+1JBZCzLNxYsbMfIeDL+lTsphD5jN5N+n0zg==", + "version": "6.21.0", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.21.0.tgz", + "integrity": "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==", "dev": true, "license": "MIT" }, diff --git a/mcp-wrapper/package.json b/mcp-wrapper/package.json index 6b94a57..2a292b1 100644 --- a/mcp-wrapper/package.json +++ b/mcp-wrapper/package.json @@ -18,8 +18,8 @@ "@modelcontextprotocol/sdk": "^1.0.0" }, "devDependencies": { - "@types/node": "^25.0.0", - "typescript": "^6.0.0", + "@types/node": "^22.0.0", + "typescript": "^5.4.0", "tsx": "^4.7.0" }, "engines": { diff --git a/mcp-wrapper/src/tools.ts b/mcp-wrapper/src/tools.ts index aa68a99..a43ed71 100644 --- a/mcp-wrapper/src/tools.ts +++ b/mcp-wrapper/src/tools.ts @@ -24,7 +24,6 @@ export const TOOL_NAMES = [ "memory_contradict", "memory_capture", "memory_consolidate", - "memory_session_context", "profile_get_set", "curiosity_pending", "schema_list", @@ -186,24 +185,6 @@ export const toolSchemas: Record = { }, }, }, - memory_session_context: { - name: "memory_session_context", - description: - "Retrieve the current session context payload (identity, session handle, " + - "topic cluster, rich club). Call at session start to load relevant " + - "background memory. Returns JSON with l0, l1, l2, rich_club, and " + - "compact_handle fields. Optional session_id parameter to query a specific session.", - inputSchema: { - type: "object", - properties: { - session_id: { - type: "string", - description: - "Optional session id to query. When omitted, uses the wrapper's current session.", - }, - }, - }, - }, profile_get_set: { name: "profile_get_set", description: @@ -367,10 +348,6 @@ export async function invokeTool( return bridge.call("memory_capture", args); case "memory_consolidate": return bridge.call("memory_consolidate", args); - case "memory_session_context": { - const sessionId = (args.session_id as string) || null; - return bridge.call("session_start_payload", sessionId ? { session_id: sessionId } : {}); - } case "profile_get_set": { const op = args.operation as string; if (op === "get") { diff --git a/src/iai_mcp/capture.py b/src/iai_mcp/capture.py index d61f4d7..67ff704 100644 --- a/src/iai_mcp/capture.py +++ b/src/iai_mcp/capture.py @@ -201,13 +201,6 @@ def capture_turn( log.exception("capture_turn insert failed") return {"status": "skipped", "record_id": None, "reason": f"insert-failed: {type(e).__name__}"} - # Wire temporal_next: link this insert to the previous same-session insert. - try: - from iai_mcp.retrieve import link_temporal_next - link_temporal_next(store, rec, session_id=session_id) - except Exception: - pass # diagnostic only; never block the write path. - return {"status": "inserted", "record_id": str(rec.id), "reason": f"tier={tier}"} diff --git a/src/iai_mcp/core.py b/src/iai_mcp/core.py index 3dfdfb9..ebaa714 100644 --- a/src/iai_mcp/core.py +++ b/src/iai_mcp/core.py @@ -741,15 +741,6 @@ def dispatch(store: MemoryStore, method: str, params: dict) -> dict: # wake_depth knob reaches the assembler. from iai_mcp.session import assemble_session_start, SessionStartPayload sid = params.get("session_id", "-") - # D5-02 per-call wake_depth override: a caller (e.g. the HTTP adapter - # serving session-context for system-prompt injection) can request - # standard/deep content WITHOUT mutating the global profile knob. - # Invalid/absent values fall back to the per-process profile state. - profile_state = _profile_state - wd_override = params.get("wake_depth") - if wd_override in ("minimal", "standard", "deep"): - profile_state = dict(_profile_state or {}) - profile_state["wake_depth"] = wd_override records_count = store.count_rows("records") if records_count == 0: empty = SessionStartPayload( @@ -765,7 +756,7 @@ def dispatch(store: MemoryStore, method: str, params: dict) -> dict: payload = assemble_session_start( store, assignment, rc, session_id=sid, - profile_state=profile_state, + profile_state=_profile_state, ) return _payload_to_json(payload) diff --git a/src/iai_mcp/daemon.py b/src/iai_mcp/daemon.py index 24ebafb..8c7fc82 100644 --- a/src/iai_mcp/daemon.py +++ b/src/iai_mcp/daemon.py @@ -1403,27 +1403,6 @@ async def main() -> int: mcp_socket = SocketServer(store, lock=lock, state=state) mcp_socket_task = asyncio.create_task(mcp_socket.serve()) - # Tier-1 PoC: optional localhost HTTP adapter over the same core.dispatch. - # OFF by default -- enabled only when IAI_DAEMON_HTTP_PORT is set (value "0" - # means OS-assigned ephemeral port, written to ~/.iai-mcp/.http.port for - # discovery). Lets MCP hosts that lack a server->client push mechanism - # (e.g. OpenCode) fetch session memory by URL. NO auth in this tier -- - # 127.0.0.1 bind is the only boundary; see http_server.py header. - http_server = None - http_task = None - _http_port_env = os.environ.get("IAI_DAEMON_HTTP_PORT") - if _http_port_env is not None: - try: - from iai_mcp.http_server import HttpServer - http_server = HttpServer(store, port=int(_http_port_env)) - http_task = asyncio.create_task(http_server.serve()) - except (ValueError, OSError): - # Bad port value or bind failure must NOT crash daemon boot - # (mirrors the maintenance try/except above). The unix socket - # and all lifecycle tasks come up regardless. - http_server = None - http_task = None - # Plan 10.6-01 Task 1.4: REMOVED `_propagate_idle_shutdown` # bridge task. The socket-side `idle_watcher` (which set # mcp_socket.shutdown_event after IDLE_CHECK_INTERVAL_SECS of @@ -1682,20 +1661,12 @@ async def main() -> int: mcp_socket.shutdown_event.set() except Exception: pass - # Drain the optional HTTP adapter the same way before cancellation. - if http_server is not None: - try: - http_server.shutdown_event.set() - except Exception: - pass _cancel_targets = [ tick_task, audit_task, s4_task, cascade_task, mcp_socket_task, cpu_watchdog_task, lifecycle_tick_task, ] - if http_task is not None: - _cancel_targets.append(http_task) for t in _cancel_targets: t.cancel() # Drain task exceptions silently: we're shutting down. diff --git a/src/iai_mcp/graph.py b/src/iai_mcp/graph.py index 4becfb5..0bba61c 100644 --- a/src/iai_mcp/graph.py +++ b/src/iai_mcp/graph.py @@ -42,7 +42,6 @@ class MemoryGraph: def __init__(self) -> None: self._nx: nx.Graph = nx.Graph() self._ig: "ig.Graph | None" = None - self._ig_stale: bool = False self._attrs: dict[UUID, dict[str, Any]] = {} self._backend: str = "networkx" @@ -80,11 +79,10 @@ class MemoryGraph: self._nx.add_edge( str(src), str(dst), weight=weight, edge_type=edge_type ) - self._maybe_switch_backend() - if self._backend == "igraph": - # Lazy rebuild: igraph mirror is stale after edge writes - # and will be rebuilt on next read. - self._ig_stale = True + if self._ig is not None: + # igraph mirror is immutable by topology; rebuild after each edge + # write while in igraph backend. Cheap enough at Phase-1 scale. + self._rebuild_igraph() # ------------------------------------------------------ backend switching @@ -107,17 +105,11 @@ class MemoryGraph: weights = [ float(self._nx[u][v].get("weight", 1.0)) for u, v in self._nx.edges() ] - g = ig.Graph(n=len(nodes), edges=edges, directed=False) # type: ignore[unbound] + g = ig.Graph(n=len(nodes), edges=edges, directed=False) g.vs["name"] = nodes if weights: g.es["weight"] = weights self._ig = g - self._ig_stale = False - - def _ensure_igraph(self) -> None: - """Rebuild igraph mirror if stale from deferred edge writes.""" - if self._ig_stale and _HAS_IGRAPH: - self._rebuild_igraph() # ---------------------------------------------------------- graph metrics @@ -132,7 +124,6 @@ class MemoryGraph: bc = nx.betweenness_centrality(self._nx, weight="weight") return {UUID(n): float(c) for n, c in bc.items()} # igraph path - self._ensure_igraph() assert self._ig is not None has_weight = "weight" in self._ig.es.attributes() raw = self._ig.betweenness(weights="weight" if has_weight else None) diff --git a/src/iai_mcp/http_server.py b/src/iai_mcp/http_server.py deleted file mode 100644 index edd9202..0000000 --- a/src/iai_mcp/http_server.py +++ /dev/null @@ -1,326 +0,0 @@ -"""Tier-1 localhost HTTP adapter over core.dispatch (PoC). - -Sibling transport to socket_server.py: both are thin adapters around the -single transport-agnostic entry point ``core.dispatch(store, method, params)`` -(D7-08 -- one dispatch function per method, no transport branching). This one -speaks HTTP/1.0 on 127.0.0.1 so an MCP host that has *no* server->client push -mechanism (e.g. OpenCode, which ignores the MCP ``instructions`` field) can -still pull session memory into its system prompt by fetching a URL. - -PoC SCOPE / NON-GOALS (deliberately minimal -- Tier 1): - - localhost-only bind. NEVER 0.0.0.0. The unix socket's security boundary is - filesystem perms (chmod 0o600, T-04-07); a TCP port has none, so we limit - blast radius to loopback and ship NO auth here. Token auth is the Tier-2 - follow-up before this is anything but a local convenience. - - one request per connection (Connection: close). No keep-alive, no chunked - encoding, no HTTP/1.1 pipelining. Simplest robust thing for fetch() clients. - - disabled unless IAI_DAEMON_HTTP_PORT is set (see daemon.main wiring). Absent - => no TCP surface at all, identical to today. - -Constitutional guards (mirror socket_server.py): - - C-DISPATCHER-FSM-ISOLATION: this adapter calls core.dispatch ONLY; it never - touches the daemon FSM. FSM transitions stay owned by daemon.py's tick. - - C3 ZERO API COST: stdlib + core.dispatch only; no SDK references. - - C5 LITERAL PRESERVATION: transport-only; zero record mutation here. - - R3 head-of-line: dispatch is sync + 50-500 ms, so it runs on - asyncio.to_thread exactly like socket_server.py:257. - - R5 fail-loud: dispatch raises map to HTTP status codes (see _STATUS_FOR). -""" - -from __future__ import annotations - -import asyncio -import json -import os -import sys -import time -from pathlib import Path -from typing import Any -from urllib.parse import parse_qs, urlsplit - -from iai_mcp.concurrency import SOCKET_PATH -from iai_mcp.core import UnknownMethodError, dispatch - -# Discovery file: consumers (OpenCode plugin) read the live port from here, -# mirroring the SOCKET_PATH convention (~/.iai-mcp/.daemon.sock). Sits next to -# the socket so both transports share the one well-known dir. -HTTP_PORT_FILE: Path = SOCKET_PATH.parent / ".http.port" - -# dispatch-exception -> HTTP status, mirroring socket_server.py's JSON-RPC code -# mapping (ERR_METHOD_NOT_FOUND/-32601, ERR_INVALID_PARAMS/-32602, ...). -_BAD_REQUEST = 400 -_NOT_FOUND = 404 -_METHOD_NOT_ALLOWED = 405 -_INTERNAL = 500 - -_MAX_BODY_BYTES = 1 << 20 # 1 MiB cap on POST bodies; refuse larger with 400. - - -def _payload_to_text(result: Any) -> str: - """Flatten a session_start_payload dict into a system-prompt-ready block. - - Best-effort and forgiving: any missing/empty key is skipped, so this is - correct at every wake_depth: - - standard/deep populate the human-readable content layers - (l0/l1/rich_club/l2) -- the useful case for system-prompt injection. - - minimal (the default) leaves those empty and emits only the compact - pointer/handle fields (~30 tok). Those are opaque references, not - content; we still surface them on a trailing line so the block is - never silently empty, but a host that wants real memory text in the - prompt must run wake_depth>=standard. - Token-accounting ints and breakpoint_marker are dropped. - """ - if not isinstance(result, dict): - return str(result) - parts: list[str] = [] - headers = {"l0": "## L0 Identity", "l1": "## L1 Recent Summary", "rich_club": "## Rich Club"} - for key in ("l0", "l1", "rich_club"): - val = result.get(key) - if isinstance(val, str) and val.strip(): - parts.append(f"{headers[key]}\n{val.strip()}") - l2 = result.get("l2") - if isinstance(l2, list): - for item in l2: - text = item if isinstance(item, str) else json.dumps(item, ensure_ascii=False) - if text.strip(): - parts.append(f"## L2 Episode\n{text.strip()}") - # Compact handles (populated at minimal; also present at standard/deep). - handles = [ - result.get(k) - for k in ("identity_pointer", "compact_handle", "brain_handle", "topic_cluster_hint") - ] - handles = [h for h in handles if isinstance(h, str) and h.strip()] - if handles: - parts.append(f"## Handles\n{' '.join(handles)}") - return "\n\n".join(parts) - - -class HttpServer: - """Per-connection HTTP/1.0 server routing to core.dispatch. - - Routes (all GET unless noted): - - GET /healthz -> {"ok": true} (no dispatch) - - GET /memory/session-context -> dispatch("session_start_payload", - {session_id?}); ?format=text returns - a flattened text/plain block. - - POST /rpc {method, params} -> dispatch(method, params); generic - passthrough mirroring the socket. - - Constructor args mirror SocketServer: a shared MemoryStore singleton plus - bind knobs. shutdown_event lets daemon.main drain gracefully. - """ - - def __init__( - self, - store: Any, - *, - host: str = "127.0.0.1", - port: int = 0, - port_file: Path | None = None, - ) -> None: - self.store = store - self.host = host - self.port = port # 0 => OS-assigned ephemeral port (written to port_file) - self.port_file = port_file if port_file is not None else HTTP_PORT_FILE - self.bound_port: int | None = None - self.last_activity_ts: float = time.monotonic() - self.active_connections: int = 0 - self.shutdown_event: asyncio.Event = asyncio.Event() - - # -- request parsing ---------------------------------------------------- - - async def _read_request( - self, reader: asyncio.StreamReader - ) -> tuple[str, str, dict[str, str], bytes] | None: - """Parse one HTTP request. Returns (method, target, headers, body) or None on EOF.""" - request_line = await reader.readline() - if not request_line: - return None - try: - method, target, _version = request_line.decode("latin-1").rstrip("\r\n").split(" ", 2) - except ValueError: - raise _HttpError(_BAD_REQUEST, "malformed request line") - headers: dict[str, str] = {} - while True: - line = await reader.readline() - if line in (b"\r\n", b"\n", b""): - break - raw = line.decode("latin-1").rstrip("\r\n") - if ":" in raw: - name, _, value = raw.partition(":") - headers[name.strip().lower()] = value.strip() - body = b"" - length = headers.get("content-length") - if length is not None: - try: - n = int(length) - except ValueError: - raise _HttpError(_BAD_REQUEST, "bad content-length") - if n > _MAX_BODY_BYTES: - raise _HttpError(_BAD_REQUEST, "body too large") - if n > 0: - body = await reader.readexactly(n) - return method.upper(), target, headers, body - - # -- routing ------------------------------------------------------------ - - async def _route(self, method: str, target: str, body: bytes) -> tuple[int, str, str]: - """Return (status, content_type, text). Raises _HttpError for 4xx/5xx shapes.""" - split = urlsplit(target) - path = split.path - query = parse_qs(split.query) - - if path == "/healthz": - if method != "GET": - raise _HttpError(_METHOD_NOT_ALLOWED, "GET only") - return 200, "application/json", json.dumps({"ok": True}) - - if path == "/memory/session-context": - if method != "GET": - raise _HttpError(_METHOD_NOT_ALLOWED, "GET only") - session_id = (query.get("session_id") or [None])[0] - params: dict = {} - if session_id: - params["session_id"] = session_id - # Per-call wake_depth override (minimal|standard|deep). standard/deep - # populate l0/l1/l2/rich_club -- the content worth injecting into a - # system prompt; core.py validates and ignores junk values. - wake_depth = (query.get("wake_depth") or [None])[0] - if wake_depth: - params["wake_depth"] = wake_depth - result = await self._dispatch("session_start_payload", params) - if (query.get("format") or ["json"])[0] == "text": - return 200, "text/plain; charset=utf-8", _payload_to_text(result) - return 200, "application/json", json.dumps(result, ensure_ascii=False) - - if path == "/rpc": - if method != "POST": - raise _HttpError(_METHOD_NOT_ALLOWED, "POST only") - try: - envelope = json.loads(body or b"{}") - except json.JSONDecodeError as e: - raise _HttpError(_BAD_REQUEST, f"invalid json: {e}") - if not isinstance(envelope, dict) or not isinstance(envelope.get("method"), str): - raise _HttpError(_BAD_REQUEST, "expected {method, params}") - result = await self._dispatch(envelope["method"], envelope.get("params") or {}) - return 200, "application/json", json.dumps(result, ensure_ascii=False) - - raise _HttpError(_NOT_FOUND, f"no route for {path}") - - async def _dispatch(self, method: str, params: dict) -> Any: - """Run core.dispatch off-loop (R3) and map raises to _HttpError (R5).""" - try: - # CRITICAL R3: dispatch is sync + 50-500 ms; to_thread prevents - # head-of-line blocking across connections (same as socket_server). - return await asyncio.to_thread(dispatch, self.store, method, params) - except UnknownMethodError as e: - raise _HttpError(_NOT_FOUND, f"unknown method '{e.args[0]}'") - except KeyError as e: - raise _HttpError(_BAD_REQUEST, f"missing required param: {e.args[0]!r}") - except TypeError as e: - raise _HttpError(_BAD_REQUEST, str(e)) - except Exception as e: # noqa: BLE001 -- HTTP must never crash daemon - raise _HttpError(_INTERNAL, str(e)) - - # -- connection handler ------------------------------------------------- - - async def handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: - """One coroutine per connection. Reads one request, responds, closes.""" - self.active_connections += 1 - t0 = time.monotonic() - method, target = "-", "-" - try: - self.last_activity_ts = t0 - try: - parsed = await self._read_request(reader) - if parsed is None: - return - method, target, _headers, body = parsed - status, content_type, text = await self._route(method, target, body) - except _HttpError as e: - status, content_type, text = e.status, "application/json", json.dumps( - {"error": e.message} - ) - except asyncio.IncompleteReadError: - status, content_type, text = _BAD_REQUEST, "application/json", json.dumps( - {"error": "truncated body"} - ) - self._write_response(writer, status, content_type, text) - await writer.drain() - # Per-request access log -> stderr (journal). Matches the daemon's - # existing {"event":...} JSON-line style so it greps cleanly. - print( - json.dumps({ - "event": "http_request", - "method": method, - "target": target, - "status": status, - "bytes": len(text.encode("utf-8")), - "ms": round((time.monotonic() - t0) * 1000), - }), - file=sys.stderr, - flush=True, - ) - except (ConnectionResetError, BrokenPipeError, ConnectionAbortedError): - # Client hung up mid-write (fetch aborted, host killed). Expected; - # not a daemon fault -- mirrors socket_server.py's handling. - pass - finally: - self.active_connections -= 1 - try: - writer.close() - await writer.wait_closed() - except Exception: - pass - - @staticmethod - def _write_response( - writer: asyncio.StreamWriter, status: int, content_type: str, text: str - ) -> None: - body = text.encode("utf-8") - reason = { - 200: "OK", 400: "Bad Request", 404: "Not Found", - 405: "Method Not Allowed", 500: "Internal Server Error", - }.get(status, "OK") - head = ( - f"HTTP/1.0 {status} {reason}\r\n" - f"Content-Type: {content_type}\r\n" - f"Content-Length: {len(body)}\r\n" - "Connection: close\r\n" - "\r\n" - ) - writer.write(head.encode("latin-1") + body) - - # -- lifecycle ---------------------------------------------------------- - - async def serve(self) -> None: - """Bind 127.0.0.1:port, write the port-file, serve until shutdown_event.""" - server = await asyncio.start_server(self.handle, host=self.host, port=self.port) - self.bound_port = server.sockets[0].getsockname()[1] - # Discovery: publish the live port so consumers needn't guess. Best - # effort -- a failed write just means callers must be told the port. - try: - self.port_file.parent.mkdir(parents=True, exist_ok=True) - self.port_file.write_text(str(self.bound_port), encoding="utf-8") - os.chmod(self.port_file, 0o600) - except OSError: - pass - try: - async with server: - await self.shutdown_event.wait() - server.close() - await server.wait_closed() - finally: - try: - self.port_file.unlink() - except (FileNotFoundError, OSError): - pass - - -class _HttpError(Exception): - """Internal control-flow exception carrying an HTTP status + message.""" - - def __init__(self, status: int, message: str) -> None: - super().__init__(message) - self.status = status - self.message = message diff --git a/src/iai_mcp/qdrant_store.py b/src/iai_mcp/qdrant_store.py index 1cc21da..99e674c 100644 --- a/src/iai_mcp/qdrant_store.py +++ b/src/iai_mcp/qdrant_store.py @@ -4,10 +4,10 @@ Replaces LanceDB with Qdrant for remote vector storage. Collections (payload-partitioned, per Qdrant best practices): - `records` : MemoryRecord rows (1024-dim cosine vectors) - All points carry `table: "records"` + `group_id` payload. + All points carry `table: "records"` + `group_id` payload. - `metadata` : Payload-only (no vectors) containing edges, events, - budget_ledger, ratelimit_ledger. - Each point carries `table` + `group_id` payload. + budget_ledger, ratelimit_ledger. + Each point carries `table` + `group_id` payload. Both collections use `is_tenant: true` payload indexes on `table` for co-located storage and per-table HNSW graphs (payload_m=16, m=0). @@ -18,18 +18,6 @@ and data_json (events). AD = record UUID bytes. The Qdrant client connects to a remote server via QDRANT_URL + QDRANT_API_KEY environment variables (overridable via constructor). - -Shim interface (_TableShim): -- ``add()`` routes by table name to dedicated upsert methods for all 5 tables - (EDGES_TABLE, EVENTS_TABLE, BUDGET_TABLE, RATELIMIT_TABLE, RECORDS_TABLE). -- ``to_pandas()`` constructs full DataFrames for edges, events, budget, and - ratelimit tables (records uses records_as_dataframe). -- ``_parse_where()`` translates LanceDB-style SQL clauses into Qdrant Filter: - ``>`` / ``>=`` / ``<`` / ``<=`` emit Range conditions; ``=`` emits - MatchValue; multiple clauses joined with AND. -- ``events_query()`` scrolls all matching events and applies Python-side - filtering so events lacking ts_epoch (pre-migration rows) are never - silently dropped. Accepts both str and datetime for the since parameter. """ from __future__ import annotations @@ -644,7 +632,6 @@ class QdrantStore: offset=offset, scroll_filter=table_filter, with_payload=True, - with_vectors=True, ) all_points.extend(points) if next_offset is None: @@ -652,64 +639,6 @@ class QdrantStore: offset = next_offset return [self._from_point(p) for p in all_points] - @staticmethod - def _parse_where(where: str | None) -> Filter | None: - """Parse a LanceDB-style where clause into a Qdrant Filter. - - Supported grammar: - - ``key = 'value'`` or ``key = "value"`` → MatchValue - - ``key > N`` or ``key >= N`` → Range(gt/gte) - - ``key < N`` or ``key <= N`` → Range(lt/lte) - - Multiple clauses joined by AND → Filter(must=[...]) - - Numeric values are auto-detected; string values must be - quoted. Returns None if the clause is empty or cannot be parsed. - """ - if not where: - return None - - clauses = re.split(r'\s+AND\s+', where, flags=re.IGNORECASE) - conditions: list = [] - - for clause in clauses: - clause = clause.strip() - if not clause: - continue - - m = re.match( - r"""([a-z_]+)\s*(>=|<=|>|<)\s*(-?\d+(?:\.\d+)?)\s*$""", - clause, - ) - if m: - key, op, val_str = m.group(1), m.group(2), m.group(3) - val = float(val_str) - if op == ">": - conditions.append(FieldCondition(key=key, range=models.Range(gt=val))) - elif op == ">=": - conditions.append(FieldCondition(key=key, range=models.Range(gte=val))) - elif op == "<": - conditions.append(FieldCondition(key=key, range=models.Range(lt=val))) - elif op == "<=": - conditions.append(FieldCondition(key=key, range=models.Range(lte=val))) - continue - - m = re.match( - r"""([a-z_]+)\s*=\s*['"]([^'"]+)['"]\s*$""", - clause, - ) - if m: - key, val = m.group(1), m.group(2) - conditions.append( - FieldCondition(key=key, match=MatchValue(value=val)) - ) - continue - - continue - - if not conditions: - return None - return Filter(must=conditions) - def iter_records( self, *, @@ -720,10 +649,11 @@ class QdrantStore: """Streaming iterator over records (filtered by table=records).""" offset = None while True: + # Build filter: always include table=records, optionally add tier conditions = [FieldCondition(key="table", match=MatchValue(value=RECORDS_TABLE))] - where_filter = self._parse_where(where) - if where_filter is not None: - conditions.extend(where_filter.must) + if where and where.startswith("tier = "): + tier = where.split("'")[1] + conditions.append(FieldCondition(key="tier", match=MatchValue(value=tier))) qdrant_filter = Filter(must=conditions) if conditions else None points, next_offset = self._client.scroll( @@ -731,7 +661,7 @@ class QdrantStore: limit=batch_size, offset=offset, scroll_filter=qdrant_filter, - with_payload=columns if columns else True, + with_payload=True, ) for point in points: yield self._from_point(point) @@ -756,9 +686,9 @@ class QdrantStore: offset = None while True: conditions = [FieldCondition(key="table", match=MatchValue(value=RECORDS_TABLE))] - where_filter = self._parse_where(where) - if where_filter is not None: - conditions.extend(where_filter.must) + if where and where.startswith("tier = "): + tier = where.split("'")[1] + conditions.append(FieldCondition(key="tier", match=MatchValue(value=tier))) qdrant_filter = Filter(must=conditions) if conditions else None points, next_offset = self._client.scroll( @@ -770,9 +700,9 @@ class QdrantStore: with_vectors=False, ) for point in points: + # Filter to requested columns row = {k: v for k, v in point.payload.items() if k in columns} - if "id" in columns: - row["id"] = point.id + row["id"] = point.id yield row if next_offset is None: @@ -1269,7 +1199,6 @@ class QdrantStore: "severity": severity, "domain": domain, "ts": ts.isoformat(), - "ts_epoch": ts.timestamp(), "data_json": data_json, "session_id": session_id, "source_ids_json": source_ids_json, @@ -1277,14 +1206,9 @@ class QdrantStore: ) self._client.upsert(collection_name=METADATA_TABLE, points=[point]) - def events_query(self, kind: str | None = None, since: str | datetime | None = None, + def events_query(self, kind: str | None = None, since: datetime | None = None, severity: str | None = None, limit: int = 100) -> list[dict]: - """Query events from the metadata collection (table=events), newest first. - - Scrolls all matching events and applies Python-side filtering so that - events lacking ``ts_epoch`` (pre-migration rows) are never silently - dropped. - """ + """Query events from the metadata collection (table=events), newest first.""" # Always filter by table=events conditions = [FieldCondition(key="table", match=MatchValue(value=EVENTS_TABLE))] if kind is not None: @@ -1293,35 +1217,16 @@ class QdrantStore: conditions.append(FieldCondition(key="severity", match=MatchValue(value=severity))) event_filter = Filter(must=conditions) if conditions else None + points, _ = self._client.scroll( + collection_name=METADATA_TABLE, + limit=limit * 10, # fetch extra for post-filtering + with_payload=True, + with_vectors=False, + scroll_filter=event_filter, + ) - # Resolve since to a datetime for Python-side comparison - since_dt: datetime | None = None - if since is not None: - if isinstance(since, str): - since_dt = datetime.fromisoformat(since) - else: - since_dt = since if since.tzinfo else since.replace(tzinfo=timezone.utc) - - # Scroll ALL matching events (no limit on scroll — limit applies after sorting) - all_points: list = [] - offset = None - while True: - points, next_offset = self._client.scroll( - collection_name=METADATA_TABLE, - limit=1000, - offset=offset, - with_payload=True, - with_vectors=False, - scroll_filter=event_filter, - ) - all_points.extend(points) - if next_offset is None: - break - offset = next_offset - - # Python-side filtering and sorting out: list[dict] = [] - for pt in all_points: + for pt in points: p = pt.payload ts_str = p.get("ts", "") try: @@ -1329,17 +1234,16 @@ class QdrantStore: except (ValueError, TypeError): ts_dt = datetime.now(timezone.utc) - # Python-side since filter (works for all events, regardless of ts_epoch) - if since_dt is not None: - cmp = ts_dt if ts_dt.tzinfo else ts_dt.replace(tzinfo=timezone.utc) - if cmp < since_dt: + if since is not None: + since_cmp = since if since.tzinfo else since.replace(tzinfo=timezone.utc) + if ts_dt < since_cmp: continue raw_data = p.get("data_json") or "{}" - # data_json is AES-GCM encrypted; pass it through as a raw string - # so query_events() in events.py can decrypt it properly. - # Pre-02-08 unencrypted rows are plain JSON — query_events() - # handles both paths via is_encrypted() / decrypt_field(). + try: + data = json.loads(raw_data) + except (TypeError, json.JSONDecodeError): + data = {} try: source_ids = json.loads(p.get("source_ids_json") or "[]") except (TypeError, json.JSONDecodeError): @@ -1351,7 +1255,7 @@ class QdrantStore: "severity": p.get("severity") or None, "domain": p.get("domain") or None, "ts": ts_dt, - "data": raw_data, + "data": data, "session_id": p.get("session_id", "-"), "source_ids": source_ids, }) @@ -1406,14 +1310,7 @@ class QdrantStore: return pd.DataFrame(columns=["src", "dst", "edge_type", "weight", "updated_at"]) def records_as_dataframe(self) -> "pd.DataFrame": - """Return all records from the records collection as a pandas DataFrame. - - Column types match the LanceDB schema: - - timestamps are datetime objects (not str) - - structure_hv is bytes (not hex string) - - provenance_json is AES-256-GCM ciphertext (not plaintext JSON) - - tags_json is a JSON string - """ + """Return all records from the records collection as a pandas DataFrame.""" try: records = self.all_records() if not records: @@ -1424,13 +1321,9 @@ class QdrantStore: "stability", "difficulty", "last_reviewed", "never_decay", "never_merge", "detail_level", "s5_trust_score", "structure_hv", - "provenance_json", "created_at", "updated_at", "schema_version", ]) rows = [] for r in records: - # Re-encrypt provenance to match LanceDB's ciphertext column. - prov_plain = json.dumps(r.provenance or []) - prov_ct = self._encrypt_for_record(r.id, prov_plain) rows.append({ "id": str(r.id), "tier": r.tier, @@ -1439,21 +1332,17 @@ class QdrantStore: "community_id": str(r.community_id) if r.community_id else None, "centrality": r.centrality, "pinned": r.pinned, - "tags_json": json.dumps(r.tags), + "tags_json": r.tags_json if hasattr(r, "tags_json") else "[]", "language": r.language, "aaak_index": r.aaak_index, "stability": r.stability, "difficulty": r.difficulty, - "last_reviewed": r.last_reviewed, + "last_reviewed": str(r.last_reviewed) if r.last_reviewed else None, "never_decay": r.never_decay, "never_merge": r.never_merge, "detail_level": r.detail_level, "s5_trust_score": r.s5_trust_score, - "structure_hv": bytes(r.structure_hv) if r.structure_hv else b"", - "provenance_json": prov_ct, - "created_at": r.created_at, - "updated_at": r.updated_at, - "schema_version": r.schema_version, + "structure_hv": r.structure_hv.hex() if r.structure_hv else "", }) return pd.DataFrame(rows) except Exception: @@ -1464,7 +1353,6 @@ class QdrantStore: "stability", "difficulty", "last_reviewed", "never_decay", "never_merge", "detail_level", "s5_trust_score", "structure_hv", - "provenance_json", "created_at", "updated_at", "schema_version", ]) # ------------------------------------------------------------------ db shim @@ -1492,100 +1380,9 @@ class QdrantStore: elif self._name == RECORDS_TABLE: return self._store.records_as_dataframe() elif self._name == EVENTS_TABLE: - return self._events_df() - elif self._name == BUDGET_TABLE: - return self._budget_df() - elif self._name == RATELIMIT_TABLE: - return self._ratelimit_df() + return pd.DataFrame() return pd.DataFrame() - def _events_df(self) -> pd.DataFrame: - """Return all events as a pandas DataFrame.""" - try: - points = self._store._scroll_all( - METADATA_TABLE, table_filter=EVENTS_TABLE, batch_size=1000, - ) - if not points: - return pd.DataFrame(columns=[ - "id", "kind", "severity", "domain", "ts", - "data_json", "session_id", "source_ids_json", - ]) - rows = [] - for pt in points: - p = pt.payload - ts_str = p.get("ts", "") - try: - ts_dt = datetime.fromisoformat(ts_str) - except (ValueError, TypeError): - ts_dt = datetime.fromisoformat("1970-01-01") - rows.append({ - "id": p.get("id", str(pt.id)), - "kind": p.get("kind", ""), - "severity": p.get("severity") or "", - "domain": p.get("domain") or "", - "ts": ts_dt, - "data_json": p.get("data_json", ""), - "session_id": p.get("session_id", "-"), - "source_ids_json": p.get("source_ids_json", "[]"), - }) - return pd.DataFrame(rows) - except Exception: - return pd.DataFrame(columns=[ - "id", "kind", "severity", "domain", "ts", - "data_json", "session_id", "source_ids_json", - ]) - - def _budget_df(self) -> pd.DataFrame: - """Return all budget_ledger rows as a pandas DataFrame.""" - try: - points = self._store._scroll_all( - METADATA_TABLE, table_filter=BUDGET_TABLE, batch_size=1000, - ) - if not points: - return pd.DataFrame(columns=["date", "usd_spent", "kind", "ts"]) - rows = [] - for pt in points: - p = pt.payload - ts_str = p.get("ts", "") - try: - ts_dt = datetime.fromisoformat(ts_str) - except (ValueError, TypeError): - ts_dt = datetime.fromisoformat("1970-01-01") - rows.append({ - "date": p.get("date", ""), - "usd_spent": float(p.get("usd_spent", 0.0)), - "kind": p.get("kind", ""), - "ts": ts_dt, - }) - return pd.DataFrame(rows) - except Exception: - return pd.DataFrame(columns=["date", "usd_spent", "kind", "ts"]) - - def _ratelimit_df(self) -> pd.DataFrame: - """Return all ratelimit_ledger rows as a pandas DataFrame.""" - try: - points = self._store._scroll_all( - METADATA_TABLE, table_filter=RATELIMIT_TABLE, batch_size=1000, - ) - if not points: - return pd.DataFrame(columns=["ts", "status_code", "endpoint"]) - rows = [] - for pt in points: - p = pt.payload - ts_str = p.get("ts", "") - try: - ts_dt = datetime.fromisoformat(ts_str) - except (ValueError, TypeError): - ts_dt = datetime.fromisoformat("1970-01-01") - rows.append({ - "ts": ts_dt, - "status_code": int(p.get("status_code", 0)), - "endpoint": p.get("endpoint", ""), - }) - return pd.DataFrame(rows) - except Exception: - return pd.DataFrame(columns=["ts", "status_code", "endpoint"]) - def delete(self, where: str) -> None: """Delete rows from the table matching the where clause. @@ -1689,206 +1486,32 @@ class QdrantStore: except Exception: pass - def add(self, rows: list[dict]) -> None: - """Insert rows into the table (LanceDB-compatible shim). + @staticmethod + def _parse_where(where: str | None) -> Filter | None: + """Parse a LanceDB-style where clause into a Qdrant Filter. - LanceDB-compatible shim: ``table.add([{"col": "val", ...}])``. - Routes to the appropriate Qdrant collection based on table name. + Supported format: ``"key = 'value' AND key2 = 'value2'"``. + Each ``key = 'value'`` segment becomes a ``FieldCondition`` + combined with ``must`` (AND semantics). + + Returns None if the clause is empty or cannot be parsed. """ - if not rows: - return - if self._name == EDGES_TABLE: - self._add_edges(rows) - elif self._name == EVENTS_TABLE: - self._add_events(rows) - elif self._name == BUDGET_TABLE: - self._add_budget(rows) - elif self._name == RATELIMIT_TABLE: - self._add_ratelimit(rows) - elif self._name == RECORDS_TABLE: - self._add_records(rows) - - def _add_edges(self, rows: list[dict]) -> None: - points = [] - for row in rows: - points.append(PointStruct( - id=str(uuid4()), - vector={}, - payload={ - "table": EDGES_TABLE, - "group_id": self._store._group_id, - "src": row.get("src", ""), - "dst": row.get("dst", ""), - "edge_type": row.get("edge_type", ""), - "weight": float(row.get("weight", 0.0)), - "updated_at": ( - row["updated_at"].isoformat() - if isinstance(row.get("updated_at"), datetime) - else str(row.get("updated_at", "")) - ), - }, - )) - if points: - self._store._client.upsert( - collection_name=METADATA_TABLE, points=points, - ) - - def _add_events(self, rows: list[dict]) -> None: - points = [] - for row in rows: - ts_val = row.get("ts") - ts_str = ( - ts_val.isoformat() - if isinstance(ts_val, datetime) - else str(ts_val) if ts_val else "" - ) - # Compute ts_epoch for numeric range filtering (matches events_add) - ts_epoch = None - if isinstance(ts_val, datetime): - ts_epoch = ts_val.timestamp() - elif ts_str: - try: - ts_epoch = datetime.fromisoformat(ts_str).timestamp() - except (ValueError, TypeError): - pass - points.append(PointStruct( - id=str(row.get("id", uuid4())), - vector={}, - payload={ - "table": EVENTS_TABLE, - "group_id": self._store._group_id, - "id": str(row.get("id", "")), - "kind": row.get("kind", ""), - "severity": row.get("severity") or "", - "domain": row.get("domain") or "", - "ts": ts_str, - "ts_epoch": ts_epoch, - "data_json": row.get("data_json", ""), - "session_id": row.get("session_id", "-"), - "source_ids_json": row.get("source_ids_json", "[]"), - }, - )) - if points: - self._store._client.upsert( - collection_name=METADATA_TABLE, points=points, - ) - - def _add_budget(self, rows: list[dict]) -> None: - points = [] - for row in rows: - ts_val = row.get("ts") - ts_str = ( - ts_val.isoformat() - if isinstance(ts_val, datetime) - else str(ts_val) if ts_val else "" - ) - points.append(PointStruct( - id=str(uuid4()), - vector={}, - payload={ - "table": BUDGET_TABLE, - "group_id": self._store._group_id, - "date": row.get("date", ""), - "usd_spent": float(row.get("usd_spent", 0.0)), - "kind": row.get("kind", ""), - "ts": ts_str, - }, - )) - if points: - self._store._client.upsert( - collection_name=METADATA_TABLE, points=points, - ) - - def _add_ratelimit(self, rows: list[dict]) -> None: - points = [] - for row in rows: - ts_val = row.get("ts") - ts_str = ( - ts_val.isoformat() - if isinstance(ts_val, datetime) - else str(ts_val) if ts_val else "" - ) - points.append(PointStruct( - id=str(uuid4()), - vector={}, - payload={ - "table": RATELIMIT_TABLE, - "group_id": self._store._group_id, - "ts": ts_str, - "status_code": int(row.get("status_code", 0)), - "endpoint": row.get("endpoint", ""), - }, - )) - if points: - self._store._client.upsert( - collection_name=METADATA_TABLE, points=points, - ) - - def _add_records(self, rows: list[dict]) -> None: - """Insert record rows — builds PointStruct directly from row dicts. - - The row dicts come from _to_row() which already contains - encrypted ciphertext for literal_surface / provenance_json / - profile_modulation_gain_json. Building a MemoryRecord and - passing it through _to_point would double-encrypt those fields, - so we construct the PointStruct payload directly here. - """ - points = [] - for row in rows: - def _ts(val): - if val is None: - return None - if isinstance(val, datetime): - return val.isoformat() - return str(val) - - # Decode structure_hv from bytes (LanceDB pa.binary → bytes) - structure_raw = row.get("structure_hv") - if isinstance(structure_raw, (bytes, bytearray)): - structure_b64 = base64.b64encode(structure_raw).decode("ascii") - elif isinstance(structure_raw, str): - # Already base64 or hex — keep as-is - structure_b64 = structure_raw - else: - structure_b64 = "" - - points.append(PointStruct( - id=str(row.get("id", uuid4())), - vector=list(row.get("embedding", [0.0] * self._store._embed_dim)), - payload={ - "table": RECORDS_TABLE, - "group_id": self._store._group_id, - "tier": row.get("tier", "episodic"), - "literal_surface": row.get("literal_surface", ""), - "aaak_index": row.get("aaak_index", ""), - "structure_hv": structure_b64, - "community_id": row.get("community_id", ""), - "centrality": float(row.get("centrality", 0.0)), - "detail_level": int(row.get("detail_level", 1)), - "pinned": bool(row.get("pinned", False)), - "stability": float(row.get("stability", 0.0)), - "difficulty": float(row.get("difficulty", 0.0)), - "last_reviewed": _ts(row.get("last_reviewed")), - "never_decay": bool(row.get("never_decay", False)), - "never_merge": bool(row.get("never_merge", False)), - "provenance_json": row.get("provenance_json", "[]"), - "created_at": _ts(row.get("created_at")), - "updated_at": _ts(row.get("updated_at")), - "tags_json": row.get("tags_json", "[]"), - "language": str(row.get("language", "en")), - "s5_trust_score": float(row.get("s5_trust_score", 0.5)), - "profile_modulation_gain_json": row.get("profile_modulation_gain_json", "{}"), - "schema_version": int(row.get("schema_version", 1)), - }, - )) - if points: - self._store._client.upsert( - collection_name=RECORDS_TABLE, points=points, - ) - - def _parse_where(self, where: str | None) -> Filter | None: - """Delegate to QdrantStore._parse_where (single source of truth).""" - return QdrantStore._parse_where(where) + if not where: + return None + # Find all key = 'value' or key = "value" patterns. + pairs = re.findall( + r"""([a-z_]+)\s*=\s*['"]([^'"]+)['"]""", + where, + ) + if not pairs: + return None + conditions = [ + FieldCondition(key=key, match=MatchValue(value=val)) + for key, val in pairs + ] + if len(conditions) == 1: + return Filter(must=conditions) + return Filter(must=conditions) @property def db(self) -> "QdrantStore._DbShim": diff --git a/src/iai_mcp/retrieve.py b/src/iai_mcp/retrieve.py index fdb46b5..72caac9 100644 --- a/src/iai_mcp/retrieve.py +++ b/src/iai_mcp/retrieve.py @@ -278,12 +278,6 @@ def contradict( store.insert(new_rec) store.add_contradicts_edge(original_id, new_rec.id) - # Wire temporal_next: contradict inserts land in same session, link them. - try: - link_temporal_next(store, new_rec, session_id="-") - except Exception: - pass # diagnostic only; never block the write path. - # monotropic proactive check fires only in high-focus # domains. Hints aren't surfaced via contradict() (its signature is fixed # to ReconsolidationReceipt), but events land in the events table so the diff --git a/src/iai_mcp/sleep.py b/src/iai_mcp/sleep.py index f2d9b1c..e3e6c5b 100644 --- a/src/iai_mcp/sleep.py +++ b/src/iai_mcp/sleep.py @@ -296,22 +296,16 @@ def run_light_consolidation( def _build_hebbian_clusters(store: MemoryStore) -> list[list[UUID]]: - """Find connected components in the hebbian+temporal_next edge graph - with size >= CLUSTER_MIN_SIZE. - - Hebbian edges capture semantic similarity (dedup/reinforce); - temporal_next edges capture sequential proximity (same-session inserts - within 5 minutes). Both are valid signals for CLS clustering. - """ + """Find connected components in the hebbian edge graph with size >= CLUSTER_MIN_SIZE.""" edges_df = store.db.open_table(EDGES_TABLE).to_pandas() if edges_df.empty: return [] - relevant = edges_df[edges_df["edge_type"].isin(("hebbian", "temporal_next"))] - if relevant.empty: + hebbian = edges_df[edges_df["edge_type"] == "hebbian"] + if hebbian.empty: return [] adj: dict[UUID, set[UUID]] = {} - for _, row in relevant.iterrows(): + for _, row in hebbian.iterrows(): src = UUID(row["src"]) dst = UUID(row["dst"]) adj.setdefault(src, set()).add(dst) diff --git a/tests/test_http_server.py b/tests/test_http_server.py deleted file mode 100644 index fdb5b02..0000000 --- a/tests/test_http_server.py +++ /dev/null @@ -1,248 +0,0 @@ -"""End-to-end tests for the Tier-1 localhost HTTP adapter (http_server.py). - -Mirrors tests/test_daemon_dispatcher.py: boot the REAL HttpServer on an -ephemeral 127.0.0.1 port and drive it with raw HTTP/1.0 over a real TCP -socket. core.dispatch is monkeypatched to a fake so these tests exercise the -*transport adapter* (routing, error->status mapping, port-file discovery, -graceful shutdown) without needing a real MemoryStore / LanceDB. -""" -from __future__ import annotations - -import asyncio -import json - -import pytest - -from iai_mcp.core import UnknownMethodError - - -def _fake_dispatch(store, method, params): - """Stand-in for core.dispatch: deterministic, no store needed.""" - if method == "session_start_payload": - return { - "l0": "pinned identity", - "l1": "recent summary", - "l2": ["episode one", "episode two"], - "rich_club": "hub concepts", - "total_cached_tokens": 0, - "total_dynamic_tokens": 1000, - # echo the override so the transport-threading test can assert it. - "wake_depth": params.get("wake_depth", "minimal"), - } - if method == "minimal_payload": - # wake_depth=minimal: content layers empty, only opaque handles set. - return { - "l0": "", "l1": "", "l2": [], "rich_club": "", - "identity_pointer": "", - "brain_handle": "", - "topic_cluster_hint": "", - "compact_handle": "", - "wake_depth": "minimal", - } - if method == "needs_param": - # Simulate core.py's params["cue"] KeyError path. - return {"cue": params["cue"]} - if method == "boom": - raise RuntimeError("kaboom") - raise UnknownMethodError(method) - - -async def _http_request(port, method, path, body=None, *, timeout=5.0): - """Send one HTTP/1.0 request, read the full response to EOF, parse it.""" - reader, writer = await asyncio.wait_for( - asyncio.open_connection("127.0.0.1", port), timeout=timeout - ) - try: - body_bytes = body.encode("utf-8") if isinstance(body, str) else (body or b"") - head = f"{method} {path} HTTP/1.0\r\n" - if body_bytes: - head += f"Content-Length: {len(body_bytes)}\r\n" - head += "Connection: close\r\n\r\n" - writer.write(head.encode("latin-1") + body_bytes) - await writer.drain() - raw = await asyncio.wait_for(reader.read(-1), timeout=timeout) - finally: - writer.close() - try: - await writer.wait_closed() - except Exception: - pass - head_bytes, _, payload = raw.partition(b"\r\n\r\n") - lines = head_bytes.split(b"\r\n") - status = int(lines[0].decode("latin-1").split(" ")[1]) - headers = {} - for line in lines[1:]: - name, _, val = line.decode("latin-1").partition(":") - headers[name.strip().lower()] = val.strip() - return status, headers, payload.decode("utf-8") - - -async def _with_http_server(coro_fn, *, port_file, monkeypatch): - """Boot the real HttpServer with _fake_dispatch on an ephemeral port.""" - from iai_mcp import http_server as hs - - monkeypatch.setattr(hs, "dispatch", _fake_dispatch) - server = hs.HttpServer(store=object(), host="127.0.0.1", port=0, port_file=port_file) - task = asyncio.create_task(server.serve()) - for _ in range(250): # wait for bind (bound_port set inside serve()) - if server.bound_port is not None: - break - await asyncio.sleep(0.01) - if server.bound_port is None: - server.shutdown_event.set() - await asyncio.wait_for(task, timeout=5) - raise AssertionError("server never bound") - try: - return await coro_fn(server) - finally: - server.shutdown_event.set() - try: - await asyncio.wait_for(task, timeout=5) - except Exception: - pass - - -def test_healthz_returns_ok(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request(server.bound_port, "GET", "/healthz") - - status, headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 200 - assert headers["content-type"] == "application/json" - assert json.loads(body) == {"ok": True} - - -def test_session_context_json(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request( - server.bound_port, "GET", "/memory/session-context?session_id=abc" - ) - - status, headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 200 - assert headers["content-type"] == "application/json" - payload = json.loads(body) - assert payload["l0"] == "pinned identity" - assert payload["l2"] == ["episode one", "episode two"] - - -def test_session_context_threads_wake_depth(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request( - server.bound_port, - "GET", - "/memory/session-context?session_id=abc&wake_depth=standard", - ) - - status, _headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 200 - assert json.loads(body)["wake_depth"] == "standard" - - -def test_session_context_text_format(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request( - server.bound_port, "GET", "/memory/session-context?format=text" - ) - - status, headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 200 - assert headers["content-type"].startswith("text/plain") - # Flattened block keeps the human layers in order, drops token ints. - assert body == ( - "## L0 Identity\npinned identity\n\n## L1 Recent Summary\nrecent summary\n\n" - "## Rich Club\nhub concepts\n\n## L2 Episode\nepisode one\n\n## L2 Episode\nepisode two" - ) - - -def test_payload_to_text_renders_handles_at_minimal(): - """At minimal wake_depth, _payload_to_text surfaces the compact handles - (content layers empty) so the injected block is never silently empty.""" - from iai_mcp.http_server import _payload_to_text - - text = _payload_to_text(_fake_dispatch(None, "minimal_payload", {})) - assert text == "## Handles\n " - - -def test_rpc_post_passthrough(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request( - server.bound_port, - "POST", - "/rpc", - json.dumps({"method": "session_start_payload", "params": {"session_id": "z"}}), - ) - - status, _headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 200 - assert json.loads(body)["rich_club"] == "hub concepts" - - -def test_unknown_route_404(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request(server.bound_port, "GET", "/nope") - - status, _headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 404 - assert "no route" in json.loads(body)["error"] - - -def test_unknown_method_maps_to_404(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request( - server.bound_port, "POST", "/rpc", json.dumps({"method": "ghost"}) - ) - - status, _headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 404 - assert "unknown method 'ghost'" in json.loads(body)["error"] - - -def test_internal_error_maps_to_500(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request( - server.bound_port, "POST", "/rpc", json.dumps({"method": "boom"}) - ) - - status, _headers, body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 500 - assert "kaboom" in json.loads(body)["error"] - - -def test_wrong_verb_405(tmp_path, monkeypatch): - async def _runner(server): - return await _http_request(server.bound_port, "POST", "/healthz") - - status, _headers, _body = asyncio.run( - _with_http_server(_runner, port_file=tmp_path / ".http.port", monkeypatch=monkeypatch) - ) - assert status == 405 - - -def test_port_file_written_and_cleaned(tmp_path, monkeypatch): - port_file = tmp_path / ".http.port" - - async def _runner(server): - # While serving, the port-file holds the live bound port. - assert port_file.read_text() == str(server.bound_port) - return server.bound_port - - asyncio.run(_with_http_server(_runner, port_file=port_file, monkeypatch=monkeypatch)) - # After graceful shutdown, the discovery file is removed. - assert not port_file.exists() diff --git a/tests/test_session_start_wake_depth.py b/tests/test_session_start_wake_depth.py deleted file mode 100644 index 4cdef16..0000000 --- a/tests/test_session_start_wake_depth.py +++ /dev/null @@ -1,58 +0,0 @@ -"""Unit tests for the per-call wake_depth override in core.dispatch. - -The HTTP adapter (and any caller) can request standard/deep session-start -content without flipping the global profile knob. These tests verify the -override is threaded into assemble_session_start's profile_state, that invalid -values fall back to the per-process state, and that the global _profile_state is -never mutated. assemble_session_start + build_runtime_graph are monkeypatched -so no real store/embedder is needed. -""" -from __future__ import annotations - -import iai_mcp.core as core -import iai_mcp.retrieve as retrieve -import iai_mcp.session as session - - -class _FakeStore: - def count_rows(self, table): # non-empty -> takes the assemble branch - return 5 - - -def _patch(monkeypatch): - captured = {} - - def fake_assemble(store, assignment, rc, *, session_id, profile_state): - captured["profile_state"] = profile_state - captured["session_id"] = session_id - return session.SessionStartPayload(l0="content") - - monkeypatch.setattr(session, "assemble_session_start", fake_assemble) - monkeypatch.setattr(retrieve, "build_runtime_graph", lambda store: (None, {}, None)) - monkeypatch.setattr(core, "_profile_state", {"wake_depth": "minimal", "literal_preservation": 0.5}) - return captured - - -def test_wake_depth_override_threaded(monkeypatch): - captured = _patch(monkeypatch) - core.dispatch(_FakeStore(), "session_start_payload", {"session_id": "s", "wake_depth": "standard"}) - ps = captured["profile_state"] - assert ps["wake_depth"] == "standard" - # other profile knobs are preserved in the override copy. - assert ps["literal_preservation"] == 0.5 - # the module global must NOT be mutated by the override. - assert core._profile_state["wake_depth"] == "minimal" - - -def test_invalid_wake_depth_falls_back_to_profile(monkeypatch): - captured = _patch(monkeypatch) - core.dispatch(_FakeStore(), "session_start_payload", {"session_id": "s", "wake_depth": "ultra"}) - # junk value ignored -> uses the per-process profile state (identity, minimal). - assert captured["profile_state"] is core._profile_state - assert captured["profile_state"]["wake_depth"] == "minimal" - - -def test_absent_wake_depth_uses_profile(monkeypatch): - captured = _patch(monkeypatch) - core.dispatch(_FakeStore(), "session_start_payload", {"session_id": "s"}) - assert captured["profile_state"] is core._profile_state