Compare commits
No commits in common. "main" and "opencode/issue8-20260522085810" have entirely different histories.
main
...
opencode/i
16 changed files with 94 additions and 1408 deletions
33
README.md
33
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`.
|
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
|
```bash
|
||||||
mkdir -p ~/.config/opencode/plugins
|
mkdir -p ~/.config/opencode/plugins
|
||||||
cp deploy/opencode/iai-mcp-capture.js ~/.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:
|
Make sure `@opencode-ai/plugin` is installed:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|
@ -157,27 +149,12 @@ cd ~/.config/opencode
|
||||||
npm install @opencode-ai/plugin
|
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
|
### Connect OpenCode
|
||||||
|
|
||||||
Add the MCP server and plugins to `~/.config/opencode/config.json`:
|
Add the MCP server to `~/.config/opencode/config.json`:
|
||||||
|
|
||||||
```json
|
```json
|
||||||
{
|
{
|
||||||
"plugin": [
|
|
||||||
"~/.config/opencode/plugins/iai-mcp-capture.js",
|
|
||||||
"~/.config/opencode/plugins/iai-mcp-memory-inject.js"
|
|
||||||
],
|
|
||||||
"mcp": {
|
"mcp": {
|
||||||
"iai-mcp": {
|
"iai-mcp": {
|
||||||
"type": "local",
|
"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
|
### Verify
|
||||||
|
|
||||||
|
|
@ -385,7 +362,7 @@ Limitations worth knowing about:
|
||||||
|
|
||||||
Claude Code is the primary host, validated in daily use.
|
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.
|
Claude Desktop should work (uses `claude_desktop_config.json` instead of `~/.claude.json`) but hasn't been tested end to end.
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -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}`);
|
|
||||||
}
|
|
||||||
},
|
|
||||||
};
|
|
||||||
};
|
|
||||||
|
|
@ -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 {}
|
|
||||||
},
|
|
||||||
};
|
|
||||||
};
|
|
||||||
33
mcp-wrapper/package-lock.json
generated
33
mcp-wrapper/package-lock.json
generated
|
|
@ -8,15 +8,16 @@
|
||||||
"name": "iai-mcp-wrapper",
|
"name": "iai-mcp-wrapper",
|
||||||
"version": "0.1.0",
|
"version": "0.1.0",
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"@modelcontextprotocol/sdk": "^1.0.0"
|
"@modelcontextprotocol/sdk": "^1.0.0",
|
||||||
|
"zod": "^4.0.0"
|
||||||
},
|
},
|
||||||
"bin": {
|
"bin": {
|
||||||
"iai-mcp-wrapper": "dist/index.js"
|
"iai-mcp-wrapper": "dist/index.js"
|
||||||
},
|
},
|
||||||
"devDependencies": {
|
"devDependencies": {
|
||||||
"@types/node": "^25.0.0",
|
"@types/node": "^22.0.0",
|
||||||
"tsx": "^4.7.0",
|
"tsx": "^4.7.0",
|
||||||
"typescript": "^6.0.0"
|
"typescript": "^5.4.0"
|
||||||
},
|
},
|
||||||
"engines": {
|
"engines": {
|
||||||
"node": ">=18"
|
"node": ">=18"
|
||||||
|
|
@ -517,13 +518,13 @@
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"node_modules/@types/node": {
|
"node_modules/@types/node": {
|
||||||
"version": "25.9.5",
|
"version": "22.19.17",
|
||||||
"resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.5.tgz",
|
"resolved": "https://registry.npmjs.org/@types/node/-/node-22.19.17.tgz",
|
||||||
"integrity": "sha512-OScDchr2fwuUmWdf4kZ9h7PcJiYDVInhJizG/biAq3cAvqwYktuy/TYGGdZNMtNTFUP7rnb0NU4TUdm82kt4Rg==",
|
"integrity": "sha512-wGdMcf+vPYM6jikpS/qhg6WiqSV/OhG+jeeHT/KlVqxYfD40iYJf9/AE1uQxVWFvU7MipKRkRv8NSHiCGgPr8Q==",
|
||||||
"dev": true,
|
"dev": true,
|
||||||
"license": "MIT",
|
"license": "MIT",
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"undici-types": ">=7.24.0 <7.24.7"
|
"undici-types": "~6.21.0"
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"node_modules/accepts": {
|
"node_modules/accepts": {
|
||||||
|
|
@ -1584,9 +1585,9 @@
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"node_modules/tsx": {
|
"node_modules/tsx": {
|
||||||
"version": "4.23.0",
|
"version": "4.22.3",
|
||||||
"resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.0.tgz",
|
"resolved": "https://registry.npmjs.org/tsx/-/tsx-4.22.3.tgz",
|
||||||
"integrity": "sha512-eUdUIaCr963q2h5u3+QwvYp0+eqPvn+egeqZUm0hwERCqqx1E3kK5ehbGCvqSE5MQAULr67ww0cA3jKc3YkM1w==",
|
"integrity": "sha512-mdoNxBC/cSQObGGVQ5Bpn5i+yv7j68gk3Nfm3wFjcJg3Z0Mix9jzAFfP12prmm5eVGmDKtp0yyArrs0Q+8gZHg==",
|
||||||
"dev": true,
|
"dev": true,
|
||||||
"license": "MIT",
|
"license": "MIT",
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
|
|
@ -1617,9 +1618,9 @@
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"node_modules/typescript": {
|
"node_modules/typescript": {
|
||||||
"version": "6.0.3",
|
"version": "5.9.3",
|
||||||
"resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz",
|
"resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz",
|
||||||
"integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==",
|
"integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==",
|
||||||
"dev": true,
|
"dev": true,
|
||||||
"license": "Apache-2.0",
|
"license": "Apache-2.0",
|
||||||
"bin": {
|
"bin": {
|
||||||
|
|
@ -1631,9 +1632,9 @@
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"node_modules/undici-types": {
|
"node_modules/undici-types": {
|
||||||
"version": "7.24.6",
|
"version": "6.21.0",
|
||||||
"resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.24.6.tgz",
|
"resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.21.0.tgz",
|
||||||
"integrity": "sha512-WRNW+sJgj5OBN4/0JpHFqtqzhpbnV0GuB+OozA9gCL7a993SmU+1JBZCzLNxYsbMfIeDL+lTsphD5jN5N+n0zg==",
|
"integrity": "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==",
|
||||||
"dev": true,
|
"dev": true,
|
||||||
"license": "MIT"
|
"license": "MIT"
|
||||||
},
|
},
|
||||||
|
|
|
||||||
|
|
@ -18,8 +18,8 @@
|
||||||
"@modelcontextprotocol/sdk": "^1.0.0"
|
"@modelcontextprotocol/sdk": "^1.0.0"
|
||||||
},
|
},
|
||||||
"devDependencies": {
|
"devDependencies": {
|
||||||
"@types/node": "^25.0.0",
|
"@types/node": "^22.0.0",
|
||||||
"typescript": "^6.0.0",
|
"typescript": "^5.4.0",
|
||||||
"tsx": "^4.7.0"
|
"tsx": "^4.7.0"
|
||||||
},
|
},
|
||||||
"engines": {
|
"engines": {
|
||||||
|
|
|
||||||
|
|
@ -24,7 +24,6 @@ export const TOOL_NAMES = [
|
||||||
"memory_contradict",
|
"memory_contradict",
|
||||||
"memory_capture",
|
"memory_capture",
|
||||||
"memory_consolidate",
|
"memory_consolidate",
|
||||||
"memory_session_context",
|
|
||||||
"profile_get_set",
|
"profile_get_set",
|
||||||
"curiosity_pending",
|
"curiosity_pending",
|
||||||
"schema_list",
|
"schema_list",
|
||||||
|
|
@ -186,24 +185,6 @@ export const toolSchemas: Record<ToolName, ToolSchema> = {
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
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: {
|
profile_get_set: {
|
||||||
name: "profile_get_set",
|
name: "profile_get_set",
|
||||||
description:
|
description:
|
||||||
|
|
@ -367,10 +348,6 @@ export async function invokeTool(
|
||||||
return bridge.call("memory_capture", args);
|
return bridge.call("memory_capture", args);
|
||||||
case "memory_consolidate":
|
case "memory_consolidate":
|
||||||
return bridge.call("memory_consolidate", args);
|
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": {
|
case "profile_get_set": {
|
||||||
const op = args.operation as string;
|
const op = args.operation as string;
|
||||||
if (op === "get") {
|
if (op === "get") {
|
||||||
|
|
|
||||||
|
|
@ -201,13 +201,6 @@ def capture_turn(
|
||||||
log.exception("capture_turn insert failed")
|
log.exception("capture_turn insert failed")
|
||||||
return {"status": "skipped", "record_id": None, "reason": f"insert-failed: {type(e).__name__}"}
|
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}"}
|
return {"status": "inserted", "record_id": str(rec.id), "reason": f"tier={tier}"}
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -741,15 +741,6 @@ def dispatch(store: MemoryStore, method: str, params: dict) -> dict:
|
||||||
# wake_depth knob reaches the assembler.
|
# wake_depth knob reaches the assembler.
|
||||||
from iai_mcp.session import assemble_session_start, SessionStartPayload
|
from iai_mcp.session import assemble_session_start, SessionStartPayload
|
||||||
sid = params.get("session_id", "-")
|
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")
|
records_count = store.count_rows("records")
|
||||||
if records_count == 0:
|
if records_count == 0:
|
||||||
empty = SessionStartPayload(
|
empty = SessionStartPayload(
|
||||||
|
|
@ -765,7 +756,7 @@ def dispatch(store: MemoryStore, method: str, params: dict) -> dict:
|
||||||
payload = assemble_session_start(
|
payload = assemble_session_start(
|
||||||
store, assignment, rc,
|
store, assignment, rc,
|
||||||
session_id=sid,
|
session_id=sid,
|
||||||
profile_state=profile_state,
|
profile_state=_profile_state,
|
||||||
)
|
)
|
||||||
return _payload_to_json(payload)
|
return _payload_to_json(payload)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1403,27 +1403,6 @@ async def main() -> int:
|
||||||
mcp_socket = SocketServer(store, lock=lock, state=state)
|
mcp_socket = SocketServer(store, lock=lock, state=state)
|
||||||
mcp_socket_task = asyncio.create_task(mcp_socket.serve())
|
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`
|
# Plan 10.6-01 Task 1.4: REMOVED `_propagate_idle_shutdown`
|
||||||
# bridge task. The socket-side `idle_watcher` (which set
|
# bridge task. The socket-side `idle_watcher` (which set
|
||||||
# mcp_socket.shutdown_event after IDLE_CHECK_INTERVAL_SECS of
|
# mcp_socket.shutdown_event after IDLE_CHECK_INTERVAL_SECS of
|
||||||
|
|
@ -1682,20 +1661,12 @@ async def main() -> int:
|
||||||
mcp_socket.shutdown_event.set()
|
mcp_socket.shutdown_event.set()
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
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 = [
|
_cancel_targets = [
|
||||||
tick_task, audit_task, s4_task, cascade_task,
|
tick_task, audit_task, s4_task, cascade_task,
|
||||||
mcp_socket_task,
|
mcp_socket_task,
|
||||||
cpu_watchdog_task,
|
cpu_watchdog_task,
|
||||||
lifecycle_tick_task,
|
lifecycle_tick_task,
|
||||||
]
|
]
|
||||||
if http_task is not None:
|
|
||||||
_cancel_targets.append(http_task)
|
|
||||||
for t in _cancel_targets:
|
for t in _cancel_targets:
|
||||||
t.cancel()
|
t.cancel()
|
||||||
# Drain task exceptions silently: we're shutting down.
|
# Drain task exceptions silently: we're shutting down.
|
||||||
|
|
|
||||||
|
|
@ -42,7 +42,6 @@ class MemoryGraph:
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
self._nx: nx.Graph = nx.Graph()
|
self._nx: nx.Graph = nx.Graph()
|
||||||
self._ig: "ig.Graph | None" = None
|
self._ig: "ig.Graph | None" = None
|
||||||
self._ig_stale: bool = False
|
|
||||||
self._attrs: dict[UUID, dict[str, Any]] = {}
|
self._attrs: dict[UUID, dict[str, Any]] = {}
|
||||||
self._backend: str = "networkx"
|
self._backend: str = "networkx"
|
||||||
|
|
||||||
|
|
@ -80,11 +79,10 @@ class MemoryGraph:
|
||||||
self._nx.add_edge(
|
self._nx.add_edge(
|
||||||
str(src), str(dst), weight=weight, edge_type=edge_type
|
str(src), str(dst), weight=weight, edge_type=edge_type
|
||||||
)
|
)
|
||||||
self._maybe_switch_backend()
|
if self._ig is not None:
|
||||||
if self._backend == "igraph":
|
# igraph mirror is immutable by topology; rebuild after each edge
|
||||||
# Lazy rebuild: igraph mirror is stale after edge writes
|
# write while in igraph backend. Cheap enough at Phase-1 scale.
|
||||||
# and will be rebuilt on next read.
|
self._rebuild_igraph()
|
||||||
self._ig_stale = True
|
|
||||||
|
|
||||||
# ------------------------------------------------------ backend switching
|
# ------------------------------------------------------ backend switching
|
||||||
|
|
||||||
|
|
@ -107,17 +105,11 @@ class MemoryGraph:
|
||||||
weights = [
|
weights = [
|
||||||
float(self._nx[u][v].get("weight", 1.0)) for u, v in self._nx.edges()
|
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
|
g.vs["name"] = nodes
|
||||||
if weights:
|
if weights:
|
||||||
g.es["weight"] = weights
|
g.es["weight"] = weights
|
||||||
self._ig = g
|
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
|
# ---------------------------------------------------------- graph metrics
|
||||||
|
|
||||||
|
|
@ -132,7 +124,6 @@ class MemoryGraph:
|
||||||
bc = nx.betweenness_centrality(self._nx, weight="weight")
|
bc = nx.betweenness_centrality(self._nx, weight="weight")
|
||||||
return {UUID(n): float(c) for n, c in bc.items()}
|
return {UUID(n): float(c) for n, c in bc.items()}
|
||||||
# igraph path
|
# igraph path
|
||||||
self._ensure_igraph()
|
|
||||||
assert self._ig is not None
|
assert self._ig is not None
|
||||||
has_weight = "weight" in self._ig.es.attributes()
|
has_weight = "weight" in self._ig.es.attributes()
|
||||||
raw = self._ig.betweenness(weights="weight" if has_weight else None)
|
raw = self._ig.betweenness(weights="weight" if has_weight else None)
|
||||||
|
|
|
||||||
|
|
@ -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
|
|
||||||
|
|
@ -4,10 +4,10 @@ Replaces LanceDB with Qdrant for remote vector storage.
|
||||||
|
|
||||||
Collections (payload-partitioned, per Qdrant best practices):
|
Collections (payload-partitioned, per Qdrant best practices):
|
||||||
- `records` : MemoryRecord rows (1024-dim cosine vectors)
|
- `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,
|
- `metadata` : Payload-only (no vectors) containing edges, events,
|
||||||
budget_ledger, ratelimit_ledger.
|
budget_ledger, ratelimit_ledger.
|
||||||
Each point carries `table` + `group_id` payload.
|
Each point carries `table` + `group_id` payload.
|
||||||
|
|
||||||
Both collections use `is_tenant: true` payload indexes on `table` for
|
Both collections use `is_tenant: true` payload indexes on `table` for
|
||||||
co-located storage and per-table HNSW graphs (payload_m=16, m=0).
|
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
|
The Qdrant client connects to a remote server via QDRANT_URL + QDRANT_API_KEY
|
||||||
environment variables (overridable via constructor).
|
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
|
from __future__ import annotations
|
||||||
|
|
||||||
|
|
@ -644,7 +632,6 @@ class QdrantStore:
|
||||||
offset=offset,
|
offset=offset,
|
||||||
scroll_filter=table_filter,
|
scroll_filter=table_filter,
|
||||||
with_payload=True,
|
with_payload=True,
|
||||||
with_vectors=True,
|
|
||||||
)
|
)
|
||||||
all_points.extend(points)
|
all_points.extend(points)
|
||||||
if next_offset is None:
|
if next_offset is None:
|
||||||
|
|
@ -652,64 +639,6 @@ class QdrantStore:
|
||||||
offset = next_offset
|
offset = next_offset
|
||||||
return [self._from_point(p) for p in all_points]
|
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(
|
def iter_records(
|
||||||
self,
|
self,
|
||||||
*,
|
*,
|
||||||
|
|
@ -720,10 +649,11 @@ class QdrantStore:
|
||||||
"""Streaming iterator over records (filtered by table=records)."""
|
"""Streaming iterator over records (filtered by table=records)."""
|
||||||
offset = None
|
offset = None
|
||||||
while True:
|
while True:
|
||||||
|
# Build filter: always include table=records, optionally add tier
|
||||||
conditions = [FieldCondition(key="table", match=MatchValue(value=RECORDS_TABLE))]
|
conditions = [FieldCondition(key="table", match=MatchValue(value=RECORDS_TABLE))]
|
||||||
where_filter = self._parse_where(where)
|
if where and where.startswith("tier = "):
|
||||||
if where_filter is not None:
|
tier = where.split("'")[1]
|
||||||
conditions.extend(where_filter.must)
|
conditions.append(FieldCondition(key="tier", match=MatchValue(value=tier)))
|
||||||
qdrant_filter = Filter(must=conditions) if conditions else None
|
qdrant_filter = Filter(must=conditions) if conditions else None
|
||||||
|
|
||||||
points, next_offset = self._client.scroll(
|
points, next_offset = self._client.scroll(
|
||||||
|
|
@ -731,7 +661,7 @@ class QdrantStore:
|
||||||
limit=batch_size,
|
limit=batch_size,
|
||||||
offset=offset,
|
offset=offset,
|
||||||
scroll_filter=qdrant_filter,
|
scroll_filter=qdrant_filter,
|
||||||
with_payload=columns if columns else True,
|
with_payload=True,
|
||||||
)
|
)
|
||||||
for point in points:
|
for point in points:
|
||||||
yield self._from_point(point)
|
yield self._from_point(point)
|
||||||
|
|
@ -756,9 +686,9 @@ class QdrantStore:
|
||||||
offset = None
|
offset = None
|
||||||
while True:
|
while True:
|
||||||
conditions = [FieldCondition(key="table", match=MatchValue(value=RECORDS_TABLE))]
|
conditions = [FieldCondition(key="table", match=MatchValue(value=RECORDS_TABLE))]
|
||||||
where_filter = self._parse_where(where)
|
if where and where.startswith("tier = "):
|
||||||
if where_filter is not None:
|
tier = where.split("'")[1]
|
||||||
conditions.extend(where_filter.must)
|
conditions.append(FieldCondition(key="tier", match=MatchValue(value=tier)))
|
||||||
qdrant_filter = Filter(must=conditions) if conditions else None
|
qdrant_filter = Filter(must=conditions) if conditions else None
|
||||||
|
|
||||||
points, next_offset = self._client.scroll(
|
points, next_offset = self._client.scroll(
|
||||||
|
|
@ -770,9 +700,9 @@ class QdrantStore:
|
||||||
with_vectors=False,
|
with_vectors=False,
|
||||||
)
|
)
|
||||||
for point in points:
|
for point in points:
|
||||||
|
# Filter to requested columns
|
||||||
row = {k: v for k, v in point.payload.items() if k in 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
|
yield row
|
||||||
|
|
||||||
if next_offset is None:
|
if next_offset is None:
|
||||||
|
|
@ -1269,7 +1199,6 @@ class QdrantStore:
|
||||||
"severity": severity,
|
"severity": severity,
|
||||||
"domain": domain,
|
"domain": domain,
|
||||||
"ts": ts.isoformat(),
|
"ts": ts.isoformat(),
|
||||||
"ts_epoch": ts.timestamp(),
|
|
||||||
"data_json": data_json,
|
"data_json": data_json,
|
||||||
"session_id": session_id,
|
"session_id": session_id,
|
||||||
"source_ids_json": source_ids_json,
|
"source_ids_json": source_ids_json,
|
||||||
|
|
@ -1277,14 +1206,9 @@ class QdrantStore:
|
||||||
)
|
)
|
||||||
self._client.upsert(collection_name=METADATA_TABLE, points=[point])
|
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]:
|
severity: str | None = None, limit: int = 100) -> list[dict]:
|
||||||
"""Query events from the metadata collection (table=events), newest first.
|
"""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.
|
|
||||||
"""
|
|
||||||
# Always filter by table=events
|
# Always filter by table=events
|
||||||
conditions = [FieldCondition(key="table", match=MatchValue(value=EVENTS_TABLE))]
|
conditions = [FieldCondition(key="table", match=MatchValue(value=EVENTS_TABLE))]
|
||||||
if kind is not None:
|
if kind is not None:
|
||||||
|
|
@ -1293,35 +1217,16 @@ class QdrantStore:
|
||||||
conditions.append(FieldCondition(key="severity", match=MatchValue(value=severity)))
|
conditions.append(FieldCondition(key="severity", match=MatchValue(value=severity)))
|
||||||
|
|
||||||
event_filter = Filter(must=conditions) if conditions else None
|
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] = []
|
out: list[dict] = []
|
||||||
for pt in all_points:
|
for pt in points:
|
||||||
p = pt.payload
|
p = pt.payload
|
||||||
ts_str = p.get("ts", "")
|
ts_str = p.get("ts", "")
|
||||||
try:
|
try:
|
||||||
|
|
@ -1329,17 +1234,16 @@ class QdrantStore:
|
||||||
except (ValueError, TypeError):
|
except (ValueError, TypeError):
|
||||||
ts_dt = datetime.now(timezone.utc)
|
ts_dt = datetime.now(timezone.utc)
|
||||||
|
|
||||||
# Python-side since filter (works for all events, regardless of ts_epoch)
|
if since is not None:
|
||||||
if since_dt is not None:
|
since_cmp = since if since.tzinfo else since.replace(tzinfo=timezone.utc)
|
||||||
cmp = ts_dt if ts_dt.tzinfo else ts_dt.replace(tzinfo=timezone.utc)
|
if ts_dt < since_cmp:
|
||||||
if cmp < since_dt:
|
|
||||||
continue
|
continue
|
||||||
|
|
||||||
raw_data = p.get("data_json") or "{}"
|
raw_data = p.get("data_json") or "{}"
|
||||||
# data_json is AES-GCM encrypted; pass it through as a raw string
|
try:
|
||||||
# so query_events() in events.py can decrypt it properly.
|
data = json.loads(raw_data)
|
||||||
# Pre-02-08 unencrypted rows are plain JSON — query_events()
|
except (TypeError, json.JSONDecodeError):
|
||||||
# handles both paths via is_encrypted() / decrypt_field().
|
data = {}
|
||||||
try:
|
try:
|
||||||
source_ids = json.loads(p.get("source_ids_json") or "[]")
|
source_ids = json.loads(p.get("source_ids_json") or "[]")
|
||||||
except (TypeError, json.JSONDecodeError):
|
except (TypeError, json.JSONDecodeError):
|
||||||
|
|
@ -1351,7 +1255,7 @@ class QdrantStore:
|
||||||
"severity": p.get("severity") or None,
|
"severity": p.get("severity") or None,
|
||||||
"domain": p.get("domain") or None,
|
"domain": p.get("domain") or None,
|
||||||
"ts": ts_dt,
|
"ts": ts_dt,
|
||||||
"data": raw_data,
|
"data": data,
|
||||||
"session_id": p.get("session_id", "-"),
|
"session_id": p.get("session_id", "-"),
|
||||||
"source_ids": source_ids,
|
"source_ids": source_ids,
|
||||||
})
|
})
|
||||||
|
|
@ -1406,14 +1310,7 @@ class QdrantStore:
|
||||||
return pd.DataFrame(columns=["src", "dst", "edge_type", "weight", "updated_at"])
|
return pd.DataFrame(columns=["src", "dst", "edge_type", "weight", "updated_at"])
|
||||||
|
|
||||||
def records_as_dataframe(self) -> "pd.DataFrame":
|
def records_as_dataframe(self) -> "pd.DataFrame":
|
||||||
"""Return all records from the records collection as a pandas 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
|
|
||||||
"""
|
|
||||||
try:
|
try:
|
||||||
records = self.all_records()
|
records = self.all_records()
|
||||||
if not records:
|
if not records:
|
||||||
|
|
@ -1424,13 +1321,9 @@ class QdrantStore:
|
||||||
"stability", "difficulty", "last_reviewed",
|
"stability", "difficulty", "last_reviewed",
|
||||||
"never_decay", "never_merge", "detail_level",
|
"never_decay", "never_merge", "detail_level",
|
||||||
"s5_trust_score", "structure_hv",
|
"s5_trust_score", "structure_hv",
|
||||||
"provenance_json", "created_at", "updated_at", "schema_version",
|
|
||||||
])
|
])
|
||||||
rows = []
|
rows = []
|
||||||
for r in records:
|
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({
|
rows.append({
|
||||||
"id": str(r.id),
|
"id": str(r.id),
|
||||||
"tier": r.tier,
|
"tier": r.tier,
|
||||||
|
|
@ -1439,21 +1332,17 @@ class QdrantStore:
|
||||||
"community_id": str(r.community_id) if r.community_id else None,
|
"community_id": str(r.community_id) if r.community_id else None,
|
||||||
"centrality": r.centrality,
|
"centrality": r.centrality,
|
||||||
"pinned": r.pinned,
|
"pinned": r.pinned,
|
||||||
"tags_json": json.dumps(r.tags),
|
"tags_json": r.tags_json if hasattr(r, "tags_json") else "[]",
|
||||||
"language": r.language,
|
"language": r.language,
|
||||||
"aaak_index": r.aaak_index,
|
"aaak_index": r.aaak_index,
|
||||||
"stability": r.stability,
|
"stability": r.stability,
|
||||||
"difficulty": r.difficulty,
|
"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_decay": r.never_decay,
|
||||||
"never_merge": r.never_merge,
|
"never_merge": r.never_merge,
|
||||||
"detail_level": r.detail_level,
|
"detail_level": r.detail_level,
|
||||||
"s5_trust_score": r.s5_trust_score,
|
"s5_trust_score": r.s5_trust_score,
|
||||||
"structure_hv": bytes(r.structure_hv) if r.structure_hv else b"",
|
"structure_hv": r.structure_hv.hex() if r.structure_hv else "",
|
||||||
"provenance_json": prov_ct,
|
|
||||||
"created_at": r.created_at,
|
|
||||||
"updated_at": r.updated_at,
|
|
||||||
"schema_version": r.schema_version,
|
|
||||||
})
|
})
|
||||||
return pd.DataFrame(rows)
|
return pd.DataFrame(rows)
|
||||||
except Exception:
|
except Exception:
|
||||||
|
|
@ -1464,7 +1353,6 @@ class QdrantStore:
|
||||||
"stability", "difficulty", "last_reviewed",
|
"stability", "difficulty", "last_reviewed",
|
||||||
"never_decay", "never_merge", "detail_level",
|
"never_decay", "never_merge", "detail_level",
|
||||||
"s5_trust_score", "structure_hv",
|
"s5_trust_score", "structure_hv",
|
||||||
"provenance_json", "created_at", "updated_at", "schema_version",
|
|
||||||
])
|
])
|
||||||
|
|
||||||
# ------------------------------------------------------------------ db shim
|
# ------------------------------------------------------------------ db shim
|
||||||
|
|
@ -1492,100 +1380,9 @@ class QdrantStore:
|
||||||
elif self._name == RECORDS_TABLE:
|
elif self._name == RECORDS_TABLE:
|
||||||
return self._store.records_as_dataframe()
|
return self._store.records_as_dataframe()
|
||||||
elif self._name == EVENTS_TABLE:
|
elif self._name == EVENTS_TABLE:
|
||||||
return self._events_df()
|
return pd.DataFrame()
|
||||||
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:
|
def delete(self, where: str) -> None:
|
||||||
"""Delete rows from the table matching the where clause.
|
"""Delete rows from the table matching the where clause.
|
||||||
|
|
||||||
|
|
@ -1689,206 +1486,32 @@ class QdrantStore:
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def add(self, rows: list[dict]) -> None:
|
@staticmethod
|
||||||
"""Insert rows into the table (LanceDB-compatible shim).
|
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", ...}])``.
|
Supported format: ``"key = 'value' AND key2 = 'value2'"``.
|
||||||
Routes to the appropriate Qdrant collection based on table name.
|
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:
|
if not where:
|
||||||
return
|
return None
|
||||||
if self._name == EDGES_TABLE:
|
# Find all key = 'value' or key = "value" patterns.
|
||||||
self._add_edges(rows)
|
pairs = re.findall(
|
||||||
elif self._name == EVENTS_TABLE:
|
r"""([a-z_]+)\s*=\s*['"]([^'"]+)['"]""",
|
||||||
self._add_events(rows)
|
where,
|
||||||
elif self._name == BUDGET_TABLE:
|
)
|
||||||
self._add_budget(rows)
|
if not pairs:
|
||||||
elif self._name == RATELIMIT_TABLE:
|
return None
|
||||||
self._add_ratelimit(rows)
|
conditions = [
|
||||||
elif self._name == RECORDS_TABLE:
|
FieldCondition(key=key, match=MatchValue(value=val))
|
||||||
self._add_records(rows)
|
for key, val in pairs
|
||||||
|
]
|
||||||
def _add_edges(self, rows: list[dict]) -> None:
|
if len(conditions) == 1:
|
||||||
points = []
|
return Filter(must=conditions)
|
||||||
for row in rows:
|
return Filter(must=conditions)
|
||||||
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)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def db(self) -> "QdrantStore._DbShim":
|
def db(self) -> "QdrantStore._DbShim":
|
||||||
|
|
|
||||||
|
|
@ -278,12 +278,6 @@ def contradict(
|
||||||
store.insert(new_rec)
|
store.insert(new_rec)
|
||||||
store.add_contradicts_edge(original_id, new_rec.id)
|
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
|
# monotropic proactive check fires only in high-focus
|
||||||
# domains. Hints aren't surfaced via contradict() (its signature is fixed
|
# domains. Hints aren't surfaced via contradict() (its signature is fixed
|
||||||
# to ReconsolidationReceipt), but events land in the events table so the
|
# to ReconsolidationReceipt), but events land in the events table so the
|
||||||
|
|
|
||||||
|
|
@ -296,22 +296,16 @@ def run_light_consolidation(
|
||||||
|
|
||||||
|
|
||||||
def _build_hebbian_clusters(store: MemoryStore) -> list[list[UUID]]:
|
def _build_hebbian_clusters(store: MemoryStore) -> list[list[UUID]]:
|
||||||
"""Find connected components in the hebbian+temporal_next edge graph
|
"""Find connected components in the hebbian edge graph with size >= CLUSTER_MIN_SIZE."""
|
||||||
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.
|
|
||||||
"""
|
|
||||||
edges_df = store.db.open_table(EDGES_TABLE).to_pandas()
|
edges_df = store.db.open_table(EDGES_TABLE).to_pandas()
|
||||||
if edges_df.empty:
|
if edges_df.empty:
|
||||||
return []
|
return []
|
||||||
relevant = edges_df[edges_df["edge_type"].isin(("hebbian", "temporal_next"))]
|
hebbian = edges_df[edges_df["edge_type"] == "hebbian"]
|
||||||
if relevant.empty:
|
if hebbian.empty:
|
||||||
return []
|
return []
|
||||||
|
|
||||||
adj: dict[UUID, set[UUID]] = {}
|
adj: dict[UUID, set[UUID]] = {}
|
||||||
for _, row in relevant.iterrows():
|
for _, row in hebbian.iterrows():
|
||||||
src = UUID(row["src"])
|
src = UUID(row["src"])
|
||||||
dst = UUID(row["dst"])
|
dst = UUID(row["dst"])
|
||||||
adj.setdefault(src, set()).add(dst)
|
adj.setdefault(src, set()).add(dst)
|
||||||
|
|
|
||||||
|
|
@ -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": "<sess:abc pend:2>",
|
|
||||||
"topic_cluster_hint": "<topic:d3256b25>",
|
|
||||||
"compact_handle": "<iai:970d1b7629da7946>",
|
|
||||||
"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<iai:970d1b7629da7946> <sess:abc pend:2> <topic:d3256b25>"
|
|
||||||
|
|
||||||
|
|
||||||
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()
|
|
||||||
|
|
@ -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
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue