From 48c690031df23d615ea03d27727fab70db5f3fe1 Mon Sep 17 00:00:00 2001 From: renovate-bot Date: Fri, 22 May 2026 12:02:04 +0000 Subject: [PATCH 01/21] chore(deps): update dependency @types/node to v22.19.19 --- mcp-wrapper/package-lock.json | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index 4213bc5..e8ced1c 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -8,8 +8,7 @@ "name": "iai-mcp-wrapper", "version": "0.1.0", "dependencies": { - "@modelcontextprotocol/sdk": "^1.0.0", - "zod": "^4.0.0" + "@modelcontextprotocol/sdk": "^1.0.0" }, "bin": { "iai-mcp-wrapper": "dist/index.js" @@ -518,9 +517,9 @@ } }, "node_modules/@types/node": { - "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==", + "version": "22.19.19", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.19.19.tgz", + "integrity": "sha512-dyh/xO2Fh5bYrfWaaqGrRQQGkNdmYw6AmaAUvYeUMNTWQtvb796ikLdmTchRmOlOiIJ1TDXfWgVx1QkUlQ6Hew==", "dev": true, "license": "MIT", "dependencies": { From a823edcc9389bd88156bd67c78986ef8dd6403c3 Mon Sep 17 00:00:00 2001 From: renovate-bot Date: Fri, 22 May 2026 12:05:57 +0000 Subject: [PATCH 02/21] chore(deps): update dependency @types/node to v25 --- mcp-wrapper/package-lock.json | 16 ++++++++-------- mcp-wrapper/package.json | 2 +- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index e8ced1c..ad53a60 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -14,7 +14,7 @@ "iai-mcp-wrapper": "dist/index.js" }, "devDependencies": { - "@types/node": "^22.0.0", + "@types/node": "^25.0.0", "tsx": "^4.7.0", "typescript": "^5.4.0" }, @@ -517,13 +517,13 @@ } }, "node_modules/@types/node": { - "version": "22.19.19", - "resolved": "https://registry.npmjs.org/@types/node/-/node-22.19.19.tgz", - "integrity": "sha512-dyh/xO2Fh5bYrfWaaqGrRQQGkNdmYw6AmaAUvYeUMNTWQtvb796ikLdmTchRmOlOiIJ1TDXfWgVx1QkUlQ6Hew==", + "version": "25.9.1", + "resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.1.tgz", + "integrity": "sha512-xfrlY7UD5rMJk3ZVJP8BNzS28J36YJg+xp+LPXV1TdWxr8uMH5A860QNxYDGQe/ylDSgjxE52Q9VnO7p75tJxg==", "dev": true, "license": "MIT", "dependencies": { - "undici-types": "~6.21.0" + "undici-types": ">=7.24.0 <7.24.7" } }, "node_modules/accepts": { @@ -1631,9 +1631,9 @@ } }, "node_modules/undici-types": { - "version": "6.21.0", - "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.21.0.tgz", - "integrity": "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==", + "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==", "dev": true, "license": "MIT" }, diff --git a/mcp-wrapper/package.json b/mcp-wrapper/package.json index 2a292b1..e3455f0 100644 --- a/mcp-wrapper/package.json +++ b/mcp-wrapper/package.json @@ -18,7 +18,7 @@ "@modelcontextprotocol/sdk": "^1.0.0" }, "devDependencies": { - "@types/node": "^22.0.0", + "@types/node": "^25.0.0", "typescript": "^5.4.0", "tsx": "^4.7.0" }, From ce34a995e127aefd29c17b103d0cc9bc32cc58ae Mon Sep 17 00:00:00 2001 From: renovate-bot Date: Fri, 22 May 2026 12:29:23 +0000 Subject: [PATCH 03/21] chore(deps): update dependency typescript to v6 --- mcp-wrapper/package-lock.json | 8 ++++---- mcp-wrapper/package.json | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index ad53a60..3d9bcdc 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -16,7 +16,7 @@ "devDependencies": { "@types/node": "^25.0.0", "tsx": "^4.7.0", - "typescript": "^5.4.0" + "typescript": "^6.0.0" }, "engines": { "node": ">=18" @@ -1617,9 +1617,9 @@ } }, "node_modules/typescript": { - "version": "5.9.3", - "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", - "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", + "version": "6.0.3", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz", + "integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==", "dev": true, "license": "Apache-2.0", "bin": { diff --git a/mcp-wrapper/package.json b/mcp-wrapper/package.json index e3455f0..6b94a57 100644 --- a/mcp-wrapper/package.json +++ b/mcp-wrapper/package.json @@ -19,7 +19,7 @@ }, "devDependencies": { "@types/node": "^25.0.0", - "typescript": "^5.4.0", + "typescript": "^6.0.0", "tsx": "^4.7.0" }, "engines": { From ebe7ebf17a663145f404873d3ed6b06b993bf624 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Mon, 25 May 2026 15:40:53 +0200 Subject: [PATCH 04/21] feat(mcp): add memory_session_context tool for opencode session recall --- mcp-wrapper/src/tools.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/mcp-wrapper/src/tools.ts b/mcp-wrapper/src/tools.ts index a43ed71..6201e74 100644 --- a/mcp-wrapper/src/tools.ts +++ b/mcp-wrapper/src/tools.ts @@ -24,6 +24,7 @@ export const TOOL_NAMES = [ "memory_contradict", "memory_capture", "memory_consolidate", + "memory_session_context", "profile_get_set", "curiosity_pending", "schema_list", @@ -185,6 +186,18 @@ 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.", + inputSchema: { + type: "object", + properties: {}, + }, + }, profile_get_set: { name: "profile_get_set", description: @@ -348,6 +361,8 @@ export async function invokeTool( return bridge.call("memory_capture", args); case "memory_consolidate": return bridge.call("memory_consolidate", args); + case "memory_session_context": + return bridge.call("session_start_payload", {}); case "profile_get_set": { const op = args.operation as string; if (op === "get") { From 31a69186a32c5dadca867e8caeee087f3cb8250f Mon Sep 17 00:00:00 2001 From: Apunkt Date: Tue, 26 May 2026 09:48:46 +0200 Subject: [PATCH 05/21] Add iai-mcp-session-init plugin for automatic memory loading in OpenCode --- README.md | 35 +++++++++++++++++----- deploy/opencode/iai-mcp-session-init.js | 40 +++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 7 deletions(-) create mode 100644 deploy/opencode/iai-mcp-session-init.js diff --git a/README.md b/README.md index f2a1b47..3b17fe2 100644 --- a/README.md +++ b/README.md @@ -133,15 +133,21 @@ 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 plugin +### Install the OpenCode plugins -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. +**Capture plugin** — 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/ ``` +**Session-init plugin** — automatically loads session background memory on every new session. The model calls `memory_session_context` before responding, so it has your identity, topic cluster, and rich club context from turn one. + +```bash +cp deploy/opencode/iai-mcp-session-init.js ~/.config/opencode/plugins/ +``` + Make sure `@opencode-ai/plugin` is installed: ```bash @@ -149,12 +155,27 @@ cd ~/.config/opencode npm install @opencode-ai/plugin ``` -### Connect OpenCode - -Add the MCP server to `~/.config/opencode/config.json`: +Register both plugins in `~/.config/opencode/config.json`: ```json { + "plugin": [ + "~/.config/opencode/plugins/iai-mcp-capture.js", + "~/.config/opencode/plugins/iai-mcp-session-init.js" + ] +} +``` + +### Connect OpenCode + +Add the MCP server and plugins to `~/.config/opencode/config.json`: + +```json +{ + "plugin": [ + "~/.config/opencode/plugins/iai-mcp-capture.js", + "~/.config/opencode/plugins/iai-mcp-session-init.js" + ], "mcp": { "iai-mcp": { "type": "local", @@ -165,7 +186,7 @@ Add the MCP server to `~/.config/opencode/config.json`: } ``` -Use the absolute path to `mcp-wrapper/dist/index.js`. The plugin file (`iai-mcp-capture.js`) must already be in `~/.config/opencode/plugins/`. +Use the absolute path to `mcp-wrapper/dist/index.js`. Both plugin files must already be in `~/.config/opencode/plugins/`. ### Verify @@ -362,7 +383,7 @@ Limitations worth knowing about: Claude Code is the primary host, validated in daily use. -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`. +OpenCode is supported via the `iai-mcp-capture.js` and `iai-mcp-session-init.js` plugins (see Install the OpenCode plugins 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-session-init.js b/deploy/opencode/iai-mcp-session-init.js new file mode 100644 index 0000000..1c49400 --- /dev/null +++ b/deploy/opencode/iai-mcp-session-init.js @@ -0,0 +1,40 @@ +/** + * iai-mcp session-init plugin for OpenCode. + * + * Sets the system prompt via chat.message hook's UserMessage.system field. + * This reaches the main agent and triggers memory_session_context on every session. + */ + +export const IaiMcpSessionInit = async ({ + client, + project, + directory, + worktree, + experimental_workspace, + serverUrl, + $, +}) => { + const injectedSessions = new Set(); + + return { + event: async ({ event }) => { + if (event.type !== "session.updated") return; + const sid = event.properties?.info?.id; + if (!sid || injectedSessions.has(sid)) return; + injectedSessions.add(sid); + }, + + "chat.message": async (input, output) => { + const sid = input.sessionID; + if (!sid || !injectedSessions.has(sid)) return; + if (injectedSessions.has(`done-${sid}`)) return; + + if (output.message) { + output.message.system = + "MANDATORY: Call `memory_session_context` (no arguments) BEFORE answering any user query."; + } + + injectedSessions.add(`done-${sid}`); + }, + }; +}; From 87c43219d7e8ba1d6ac58931f7d9d78ee794ba73 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Wed, 27 May 2026 13:03:15 +0200 Subject: [PATCH 06/21] =?UTF-8?q?fix(qdrant=5Fstore):=20close=20all=207=20?= =?UTF-8?q?LanceDB=E2=86=94Qdrant=20implementation=20gaps?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _TableShim.add(): route by table name to dedicated upsert methods for all 5 tables - _TableShim.to_pandas(): full DataFrame construction for edges/events/budget/ratelimit - _parse_where: full parser with >, >=, <, <= (→Range), = (→MatchValue), AND splitting - records_as_dataframe: include provenance_json, created_at, schema_version columns - iter_records: wire columns parameter to Qdrant with_payload - iter_record_columns: stop forcing 'id' into every yielded row - events_query: scroll all events + Python-side since filter (handles missing ts_epoch) - events_add: store ts_epoch alongside ts for numeric range filtering - Module docstring: document _TableShim interface --- src/iai_mcp/qdrant_store.py | 404 ++++++++++++++++++++++++++++++++---- 1 file changed, 369 insertions(+), 35 deletions(-) diff --git a/src/iai_mcp/qdrant_store.py b/src/iai_mcp/qdrant_store.py index 99e674c..8668535 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,6 +18,18 @@ 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 @@ -661,7 +673,7 @@ class QdrantStore: limit=batch_size, offset=offset, scroll_filter=qdrant_filter, - with_payload=True, + with_payload=columns if columns else True, ) for point in points: yield self._from_point(point) @@ -702,7 +714,8 @@ class QdrantStore: for point in points: # Filter to requested columns row = {k: v for k, v in point.payload.items() if k in columns} - row["id"] = point.id + if "id" in columns: + row["id"] = point.id yield row if next_offset is None: @@ -1199,6 +1212,7 @@ 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, @@ -1206,9 +1220,14 @@ class QdrantStore: ) self._client.upsert(collection_name=METADATA_TABLE, points=[point]) - def events_query(self, kind: str | None = None, since: datetime | None = None, + def events_query(self, kind: str | None = None, since: str | datetime | None = None, 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 conditions = [FieldCondition(key="table", match=MatchValue(value=EVENTS_TABLE))] if kind is not None: @@ -1217,16 +1236,35 @@ 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 points: + for pt in all_points: p = pt.payload ts_str = p.get("ts", "") try: @@ -1234,9 +1272,10 @@ class QdrantStore: except (ValueError, TypeError): ts_dt = datetime.now(timezone.utc) - if since is not None: - since_cmp = since if since.tzinfo else since.replace(tzinfo=timezone.utc) - if ts_dt < since_cmp: + # 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: continue raw_data = p.get("data_json") or "{}" @@ -1321,6 +1360,7 @@ class QdrantStore: "stability", "difficulty", "last_reviewed", "never_decay", "never_merge", "detail_level", "s5_trust_score", "structure_hv", + "provenance_json", "created_at", "schema_version", ]) rows = [] for r in records: @@ -1343,6 +1383,9 @@ class QdrantStore: "detail_level": r.detail_level, "s5_trust_score": r.s5_trust_score, "structure_hv": r.structure_hv.hex() if r.structure_hv else "", + "provenance_json": json.dumps(r.provenance) if r.provenance else "[]", + "created_at": str(r.created_at) if r.created_at else None, + "schema_version": r.schema_version, }) return pd.DataFrame(rows) except Exception: @@ -1353,6 +1396,7 @@ class QdrantStore: "stability", "difficulty", "last_reviewed", "never_decay", "never_merge", "detail_level", "s5_trust_score", "structure_hv", + "provenance_json", "created_at", "schema_version", ]) # ------------------------------------------------------------------ db shim @@ -1380,9 +1424,100 @@ class QdrantStore: elif self._name == RECORDS_TABLE: return self._store.records_as_dataframe() elif self._name == EVENTS_TABLE: - return pd.DataFrame() + 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() + 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. @@ -1486,29 +1621,228 @@ class QdrantStore: except Exception: pass - @staticmethod - def _parse_where(where: str | None) -> Filter | None: + def add(self, rows: list[dict]) -> None: + """Insert rows into the table (LanceDB-compatible shim). + + LanceDB-compatible shim: ``table.add([{"col": "val", ...}])``. + Routes to the appropriate Qdrant collection based on table name. + """ + 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 "" + ) + 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, + "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 — converts dicts to PointStruct via _to_point.""" + points = [] + for row in rows: + # Build a minimal MemoryRecord from the dict so _to_point works + from uuid import UUID as _UUID + rec = MemoryRecord( + id=_UUID(row["id"]), + tier=row.get("tier", "episodic"), + literal_surface="", + aaak_index=row.get("aaak_index", ""), + embedding=list(row.get("embedding", [0.0] * self._store._embed_dim)), + community_id=_UUID(row["community_id"]) if row.get("community_id") else None, + 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=None, + never_decay=bool(row.get("never_decay", False)), + never_merge=bool(row.get("never_merge", False)), + provenance=[], + created_at=None, + updated_at=None, + tags=json.loads(row.get("tags_json") or "[]"), + language=row.get("language", "en"), + s5_trust_score=float(row.get("s5_trust_score", 0.5)), + profile_modulation_gain={}, + schema_version=int(row.get("schema_version", 1)), + structure_hv=b"", + ) + points.append(self._store._to_point(rec)) + if points: + self._store._client.upsert( + collection_name=RECORDS_TABLE, points=points, + ) + + def _parse_where(self, where: str | None) -> Filter | None: """Parse a LanceDB-style where clause into a Qdrant Filter. - Supported format: ``"key = 'value' AND key2 = 'value2'"``. - Each ``key = 'value'`` segment becomes a ``FieldCondition`` - combined with ``must`` (AND semantics). + 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=[...]) - Returns None if the clause is empty or cannot be parsed. + 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 - # Find all key = 'value' or key = "value" patterns. - pairs = re.findall( - r"""([a-z_]+)\s*=\s*['"]([^'"]+)['"]""", - where, - ) - if not pairs: + + # Split on AND (case-insensitive) to get individual clauses. + # Use regex to avoid splitting inside quoted strings. + clauses = re.split(r'\s+AND\s+', where, flags=re.IGNORECASE) + conditions: list = [] + + for clause in clauses: + clause = clause.strip() + if not clause: + continue + + # Try numeric comparison operators first: >, >=, <, <= + 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 + + # Try string equality: key = 'value' or key = "value" + 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 + + # Unparseable clause — skip silently (LanceDB would too). + continue + + if not conditions: 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) From a6d6e46b4cee6848a447f4ffc2b25cc8aa1b3132 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Wed, 27 May 2026 16:58:50 +0200 Subject: [PATCH 07/21] Simplify session-init: only send SDK prompt, accept session title tradeoff --- deploy/opencode/iai-mcp-session-init.js | 32 ++++++++++++------------- 1 file changed, 15 insertions(+), 17 deletions(-) diff --git a/deploy/opencode/iai-mcp-session-init.js b/deploy/opencode/iai-mcp-session-init.js index 1c49400..07adfbf 100644 --- a/deploy/opencode/iai-mcp-session-init.js +++ b/deploy/opencode/iai-mcp-session-init.js @@ -1,8 +1,8 @@ /** * iai-mcp session-init plugin for OpenCode. * - * Sets the system prompt via chat.message hook's UserMessage.system field. - * This reaches the main agent and triggers memory_session_context on every session. + * Sends SDK prompt on session.updated to inject the system instruction. + * The model calls memory_session_context on every new session. */ export const IaiMcpSessionInit = async ({ @@ -14,27 +14,25 @@ export const IaiMcpSessionInit = async ({ serverUrl, $, }) => { - const injectedSessions = new Set(); + const doneSessions = new Set(); return { event: async ({ event }) => { if (event.type !== "session.updated") return; const sid = event.properties?.info?.id; - if (!sid || injectedSessions.has(sid)) return; - injectedSessions.add(sid); - }, + if (!sid || doneSessions.has(sid)) return; + doneSessions.add(sid); - "chat.message": async (input, output) => { - const sid = input.sessionID; - if (!sid || !injectedSessions.has(sid)) return; - if (injectedSessions.has(`done-${sid}`)) return; - - if (output.message) { - output.message.system = - "MANDATORY: Call `memory_session_context` (no arguments) BEFORE answering any user query."; - } - - injectedSessions.add(`done-${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 {} }, }; }; From 4ba8e3fc19d6edcf751e64487332799afb033bef Mon Sep 17 00:00:00 2001 From: Apunkt Date: Wed, 27 May 2026 16:59:11 +0200 Subject: [PATCH 08/21] Document SDK message tradeoff in README --- README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/README.md b/README.md index 3b17fe2..1c4454f 100644 --- a/README.md +++ b/README.md @@ -148,6 +148,8 @@ cp deploy/opencode/iai-mcp-capture.js ~/.config/opencode/plugins/ cp deploy/opencode/iai-mcp-session-init.js ~/.config/opencode/plugins/ ``` +Note: a brief "INIT" message will appear in chat on the first turn. This is the mechanism that delivers the system instruction to the model. + Make sure `@opencode-ai/plugin` is installed: ```bash From 37ee97b8ec56756564a526af27ecb2c2289d64c8 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Wed, 27 May 2026 18:26:10 +0200 Subject: [PATCH 09/21] Add session_id param to memory_session_context tool; finalize session-init plugin --- mcp-wrapper/src/tools.ts | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/mcp-wrapper/src/tools.ts b/mcp-wrapper/src/tools.ts index 6201e74..aa68a99 100644 --- a/mcp-wrapper/src/tools.ts +++ b/mcp-wrapper/src/tools.ts @@ -192,10 +192,16 @@ export const toolSchemas: Record = { "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.", + "compact_handle fields. Optional session_id parameter to query a specific session.", inputSchema: { type: "object", - properties: {}, + properties: { + session_id: { + type: "string", + description: + "Optional session id to query. When omitted, uses the wrapper's current session.", + }, + }, }, }, profile_get_set: { @@ -361,8 +367,10 @@ export async function invokeTool( return bridge.call("memory_capture", args); case "memory_consolidate": return bridge.call("memory_consolidate", args); - case "memory_session_context": - return bridge.call("session_start_payload", {}); + 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") { From 5da28062b95ec6c9583ecec1b1b292d8405bb3b4 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Wed, 27 May 2026 18:50:08 +0200 Subject: [PATCH 10/21] feat: works, but no title gen in opencode --- src/iai_mcp/qdrant_store.py | 256 +++++++++++++++++++++--------------- 1 file changed, 149 insertions(+), 107 deletions(-) diff --git a/src/iai_mcp/qdrant_store.py b/src/iai_mcp/qdrant_store.py index 8668535..9e1cc63 100644 --- a/src/iai_mcp/qdrant_store.py +++ b/src/iai_mcp/qdrant_store.py @@ -651,6 +651,64 @@ 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, *, @@ -661,11 +719,10 @@ 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))] - if where and where.startswith("tier = "): - tier = where.split("'")[1] - conditions.append(FieldCondition(key="tier", match=MatchValue(value=tier))) + where_filter = self._parse_where(where) + if where_filter is not None: + conditions.extend(where_filter.must) qdrant_filter = Filter(must=conditions) if conditions else None points, next_offset = self._client.scroll( @@ -698,9 +755,9 @@ class QdrantStore: offset = None while True: conditions = [FieldCondition(key="table", match=MatchValue(value=RECORDS_TABLE))] - if where and where.startswith("tier = "): - tier = where.split("'")[1] - conditions.append(FieldCondition(key="tier", match=MatchValue(value=tier))) + where_filter = self._parse_where(where) + if where_filter is not None: + conditions.extend(where_filter.must) qdrant_filter = Filter(must=conditions) if conditions else None points, next_offset = self._client.scroll( @@ -712,7 +769,6 @@ 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 @@ -1349,7 +1405,14 @@ 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.""" + """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: records = self.all_records() if not records: @@ -1360,10 +1423,13 @@ class QdrantStore: "stability", "difficulty", "last_reviewed", "never_decay", "never_merge", "detail_level", "s5_trust_score", "structure_hv", - "provenance_json", "created_at", "schema_version", + "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, @@ -1372,19 +1438,20 @@ class QdrantStore: "community_id": str(r.community_id) if r.community_id else None, "centrality": r.centrality, "pinned": r.pinned, - "tags_json": r.tags_json if hasattr(r, "tags_json") else "[]", + "tags_json": json.dumps(r.tags), "language": r.language, "aaak_index": r.aaak_index, "stability": r.stability, "difficulty": r.difficulty, - "last_reviewed": str(r.last_reviewed) if r.last_reviewed else None, + "last_reviewed": r.last_reviewed, "never_decay": r.never_decay, "never_merge": r.never_merge, "detail_level": r.detail_level, "s5_trust_score": r.s5_trust_score, - "structure_hv": r.structure_hv.hex() if r.structure_hv else "", - "provenance_json": json.dumps(r.provenance) if r.provenance else "[]", - "created_at": str(r.created_at) if r.created_at else None, + "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, }) return pd.DataFrame(rows) @@ -1396,7 +1463,7 @@ class QdrantStore: "stability", "difficulty", "last_reviewed", "never_decay", "never_merge", "detail_level", "s5_trust_score", "structure_hv", - "provenance_json", "created_at", "schema_version", + "provenance_json", "created_at", "updated_at", "schema_version", ]) # ------------------------------------------------------------------ db shim @@ -1674,6 +1741,15 @@ class QdrantStore: 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={}, @@ -1685,6 +1761,7 @@ class QdrantStore: "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", "[]"), @@ -1747,105 +1824,70 @@ class QdrantStore: ) def _add_records(self, rows: list[dict]) -> None: - """Insert record rows — converts dicts to PointStruct via _to_point.""" + """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: - # Build a minimal MemoryRecord from the dict so _to_point works - from uuid import UUID as _UUID - rec = MemoryRecord( - id=_UUID(row["id"]), - tier=row.get("tier", "episodic"), - literal_surface="", - aaak_index=row.get("aaak_index", ""), - embedding=list(row.get("embedding", [0.0] * self._store._embed_dim)), - community_id=_UUID(row["community_id"]) if row.get("community_id") else None, - 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=None, - never_decay=bool(row.get("never_decay", False)), - never_merge=bool(row.get("never_merge", False)), - provenance=[], - created_at=None, - updated_at=None, - tags=json.loads(row.get("tags_json") or "[]"), - language=row.get("language", "en"), - s5_trust_score=float(row.get("s5_trust_score", 0.5)), - profile_modulation_gain={}, - schema_version=int(row.get("schema_version", 1)), - structure_hv=b"", - ) - points.append(self._store._to_point(rec)) + 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: - """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 - - # Split on AND (case-insensitive) to get individual clauses. - # Use regex to avoid splitting inside quoted strings. - clauses = re.split(r'\s+AND\s+', where, flags=re.IGNORECASE) - conditions: list = [] - - for clause in clauses: - clause = clause.strip() - if not clause: - continue - - # Try numeric comparison operators first: >, >=, <, <= - 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 - - # Try string equality: key = 'value' or key = "value" - 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 - - # Unparseable clause — skip silently (LanceDB would too). - continue - - if not conditions: - return None - if len(conditions) == 1: - return Filter(must=conditions) - return Filter(must=conditions) + """Delegate to QdrantStore._parse_where (single source of truth).""" + return QdrantStore._parse_where(where) @property def db(self) -> "QdrantStore._DbShim": From 4f6b91aef82e237083749c8a38a3fe921a0e6db2 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Thu, 28 May 2026 10:38:10 +0200 Subject: [PATCH 11/21] feat: adding http-mcp and opencode plugin extension for seamless integration --- deploy/opencode/iai-mcp-memory-inject.js | 111 ++++++++ src/iai_mcp/core.py | 11 +- src/iai_mcp/daemon.py | 29 ++ src/iai_mcp/http_server.py | 325 +++++++++++++++++++++++ tests/test_http_server.py | 247 +++++++++++++++++ tests/test_session_start_wake_depth.py | 58 ++++ 6 files changed, 780 insertions(+), 1 deletion(-) create mode 100644 deploy/opencode/iai-mcp-memory-inject.js create mode 100644 src/iai_mcp/http_server.py create mode 100644 tests/test_http_server.py create mode 100644 tests/test_session_start_wake_depth.py diff --git a/deploy/opencode/iai-mcp-memory-inject.js b/deploy/opencode/iai-mcp-memory-inject.js new file mode 100644 index 0000000..a1c47a3 --- /dev/null +++ b/deploy/opencode/iai-mcp-memory-inject.js @@ -0,0 +1,111 @@ +/** + * 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 failed fetches per session + +const memo = new Map(); // sessionID -> injected text (cached on success) +const attempts = new Map(); // sessionID -> failed-fetch count +const inflight = new Set(); // sessionIDs with a fetch in flight (dedupe warm+inject) + +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 ""; + 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) }); + if (!res.ok) return ""; + return (await res.text()).trim(); + } catch { + return ""; + } +} + +// 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. +async function ensureMemory(sessionId) { + const cached = memo.get(sessionId); + if (cached !== undefined) return cached; + if (inflight.has(sessionId)) return ""; // a fetch is already running + if ((attempts.get(sessionId) || 0) >= MAX_ATTEMPTS) return ""; + inflight.add(sessionId); + try { + const text = await fetchMemory(sessionId); + if (text) memo.set(sessionId, text); + else attempts.set(sessionId, (attempts.get(sessionId) || 0) + 1); + return text; + } 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; + const text = await ensureMemory(sid); + if (text) { + 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/src/iai_mcp/core.py b/src/iai_mcp/core.py index ebaa714..3dfdfb9 100644 --- a/src/iai_mcp/core.py +++ b/src/iai_mcp/core.py @@ -741,6 +741,15 @@ 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( @@ -756,7 +765,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 8c7fc82..24ebafb 100644 --- a/src/iai_mcp/daemon.py +++ b/src/iai_mcp/daemon.py @@ -1403,6 +1403,27 @@ 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 @@ -1661,12 +1682,20 @@ 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/http_server.py b/src/iai_mcp/http_server.py new file mode 100644 index 0000000..19a72ed --- /dev/null +++ b/src/iai_mcp/http_server.py @@ -0,0 +1,325 @@ +"""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] = [] + for key in ("l0", "l1", "rich_club"): + val = result.get(key) + if isinstance(val, str) and val.strip(): + parts.append(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(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(" ".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/tests/test_http_server.py b/tests/test_http_server.py new file mode 100644 index 0000000..06da13c --- /dev/null +++ b/tests/test_http_server.py @@ -0,0 +1,247 @@ +"""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 == ( + "pinned identity\n\nrecent summary\n\nhub concepts\n\nepisode one\n\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 == " " + + +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 new file mode 100644 index 0000000..4cdef16 --- /dev/null +++ b/tests/test_session_start_wake_depth.py @@ -0,0 +1,58 @@ +"""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 From 7559bac57f8e615c66051bbe29d37f3d4f6c399d Mon Sep 17 00:00:00 2001 From: Apunkt Date: Wed, 3 Jun 2026 16:47:55 +0200 Subject: [PATCH 12/21] fix(opencode): fix memory-inject race condition, empty-memory ban, add turn limit Three critical fixes to the iai-mcp-memory-inject.js plugin: - Race condition: inflight is now a Map so concurrent callers (warm-on-create event + first transform hook) await the same fetch instead of getting empty string. First turn now gets memory. - Empty-memory ban: fetchMemory returns {ok, text} tuple. HTTP 200 with empty body is memoized as valid (new user, minimal mode). Only actual network errors count against the retry budget. - Turn-limited injection: new INJECT_MAX_TURNS (default 3) stops injecting after N turns. Memory was already in system prompt; the model has seen it. Controls per-turn token cost that was 50x the old approach. Also: README now documents memory-inject.js (not session-init), and session-init.js is marked @deprecated. --- README.md | 12 ++--- deploy/opencode/iai-mcp-memory-inject.js | 66 ++++++++++++++++++------ deploy/opencode/iai-mcp-session-init.js | 15 ++++-- 3 files changed, 69 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index 1c4454f..62fb52c 100644 --- a/README.md +++ b/README.md @@ -142,13 +142,13 @@ mkdir -p ~/.config/opencode/plugins cp deploy/opencode/iai-mcp-capture.js ~/.config/opencode/plugins/ ``` -**Session-init plugin** — automatically loads session background memory on every new session. The model calls `memory_session_context` before responding, so it has your identity, topic cluster, and rich club context from turn one. +**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-session-init.js ~/.config/opencode/plugins/ +cp deploy/opencode/iai-mcp-memory-inject.js ~/.config/opencode/plugins/ ``` -Note: a brief "INIT" message will appear in chat on the first turn. This is the mechanism that delivers the system instruction to the model. +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: @@ -163,7 +163,7 @@ Register both plugins in `~/.config/opencode/config.json`: { "plugin": [ "~/.config/opencode/plugins/iai-mcp-capture.js", - "~/.config/opencode/plugins/iai-mcp-session-init.js" + "~/.config/opencode/plugins/iai-mcp-memory-inject.js" ] } ``` @@ -176,7 +176,7 @@ Add the MCP server and plugins to `~/.config/opencode/config.json`: { "plugin": [ "~/.config/opencode/plugins/iai-mcp-capture.js", - "~/.config/opencode/plugins/iai-mcp-session-init.js" + "~/.config/opencode/plugins/iai-mcp-memory-inject.js" ], "mcp": { "iai-mcp": { @@ -385,7 +385,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-session-init.js` plugins (see Install the OpenCode plugins above) and MCP server config in `~/.config/opencode/config.json`. +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). 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 index a1c47a3..81162e4 100644 --- a/deploy/opencode/iai-mcp-memory-inject.js +++ b/deploy/opencode/iai-mcp-memory-inject.js @@ -26,11 +26,13 @@ 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 failed fetches per session +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 (cached on success) -const attempts = new Map(); // sessionID -> failed-fetch count -const inflight = new Set(); // sessionIDs with a fetch in flight (dedupe warm+inject) +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"); @@ -44,17 +46,19 @@ async function readPort() { async function fetchMemory(sessionId) { const port = await readPort(); - if (!port) return ""; + 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) }); - if (!res.ok) return ""; - return (await res.text()).trim(); + 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 ""; + return { ok: false, text: "" }; } } @@ -62,17 +66,41 @@ async function fetchMemory(sessionId) { // 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; - if (inflight.has(sessionId)) return ""; // a fetch is already running - if ((attempts.get(sessionId) || 0) >= MAX_ATTEMPTS) return ""; - inflight.add(sessionId); - try { - const text = await fetchMemory(sessionId); - if (text) memo.set(sessionId, text); - else attempts.set(sessionId, (attempts.get(sessionId) || 0) + 1); + + // 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); } @@ -98,8 +126,16 @@ export const IaiMcpMemoryInject = async () => { 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) { diff --git a/deploy/opencode/iai-mcp-session-init.js b/deploy/opencode/iai-mcp-session-init.js index 07adfbf..0949d76 100644 --- a/deploy/opencode/iai-mcp-session-init.js +++ b/deploy/opencode/iai-mcp-session-init.js @@ -1,8 +1,17 @@ /** - * iai-mcp session-init plugin for OpenCode. + * @deprecated Use `iai-mcp-memory-inject.js` instead. * - * Sends SDK prompt on session.updated to inject the system instruction. - * The model calls memory_session_context on every new session. + * 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 ({ From eec312dcb41f757c43d642c5353b75a37d0756c5 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Wed, 3 Jun 2026 16:50:02 +0200 Subject: [PATCH 13/21] feat(http_server): add section headers to _payload_to_text for model parsing The flattened text output now uses markdown headers so the model can distinguish L0 identity from L1 summary from L2 episodes from handles. This replaces undifferentiated newline-separated text with structured sections: ## L0 Identity, ## L1 Recent Summary, ## Rich Club, ## L2 Episode, ## Handles. --- src/iai_mcp/http_server.py | 7 ++++--- tests/test_http_server.py | 5 +++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/src/iai_mcp/http_server.py b/src/iai_mcp/http_server.py index 19a72ed..edd9202 100644 --- a/src/iai_mcp/http_server.py +++ b/src/iai_mcp/http_server.py @@ -73,16 +73,17 @@ def _payload_to_text(result: Any) -> str: 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(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(text.strip()) + parts.append(f"## L2 Episode\n{text.strip()}") # Compact handles (populated at minimal; also present at standard/deep). handles = [ result.get(k) @@ -90,7 +91,7 @@ def _payload_to_text(result: Any) -> str: ] handles = [h for h in handles if isinstance(h, str) and h.strip()] if handles: - parts.append(" ".join(handles)) + parts.append(f"## Handles\n{' '.join(handles)}") return "\n\n".join(parts) diff --git a/tests/test_http_server.py b/tests/test_http_server.py index 06da13c..fdb5b02 100644 --- a/tests/test_http_server.py +++ b/tests/test_http_server.py @@ -158,7 +158,8 @@ def test_session_context_text_format(tmp_path, monkeypatch): assert headers["content-type"].startswith("text/plain") # Flattened block keeps the human layers in order, drops token ints. assert body == ( - "pinned identity\n\nrecent summary\n\nhub concepts\n\nepisode one\n\nepisode two" + "## 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" ) @@ -168,7 +169,7 @@ def test_payload_to_text_renders_handles_at_minimal(): from iai_mcp.http_server import _payload_to_text text = _payload_to_text(_fake_dispatch(None, "minimal_payload", {})) - assert text == " " + assert text == "## Handles\n " def test_rpc_post_passthrough(tmp_path, monkeypatch): From 344e5edaa87f1f73d2f6af520b8142a66767db3a Mon Sep 17 00:00:00 2001 From: Apunkt Date: Thu, 11 Jun 2026 12:18:03 +0200 Subject: [PATCH 14/21] Fix temporal_next edges: wire link_temporal_next into capture path, fix events_query decryption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - capture.py: call link_temporal_next() after store.insert() in capture_turn() - retrieve.py: call link_temporal_next() in contradict() for contradict inserts - qdrant_store.py: events_query() was json.loads() on encrypted data_json, losing record_id — pass raw string so query_events() can decrypt it --- src/iai_mcp/capture.py | 7 +++++++ src/iai_mcp/qdrant_store.py | 10 +++++----- src/iai_mcp/retrieve.py | 6 ++++++ 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/src/iai_mcp/capture.py b/src/iai_mcp/capture.py index 67ff704..d61f4d7 100644 --- a/src/iai_mcp/capture.py +++ b/src/iai_mcp/capture.py @@ -201,6 +201,13 @@ 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/qdrant_store.py b/src/iai_mcp/qdrant_store.py index 9e1cc63..ec6222d 100644 --- a/src/iai_mcp/qdrant_store.py +++ b/src/iai_mcp/qdrant_store.py @@ -1335,10 +1335,10 @@ class QdrantStore: continue raw_data = p.get("data_json") or "{}" - try: - data = json.loads(raw_data) - except (TypeError, json.JSONDecodeError): - data = {} + # 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: source_ids = json.loads(p.get("source_ids_json") or "[]") except (TypeError, json.JSONDecodeError): @@ -1350,7 +1350,7 @@ class QdrantStore: "severity": p.get("severity") or None, "domain": p.get("domain") or None, "ts": ts_dt, - "data": data, + "data": raw_data, "session_id": p.get("session_id", "-"), "source_ids": source_ids, }) diff --git a/src/iai_mcp/retrieve.py b/src/iai_mcp/retrieve.py index 72caac9..fdb46b5 100644 --- a/src/iai_mcp/retrieve.py +++ b/src/iai_mcp/retrieve.py @@ -278,6 +278,12 @@ 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 From 1b7c89de3bd24a4195f7bbb428a6144360b3e45d Mon Sep 17 00:00:00 2001 From: Apunkt Date: Fri, 12 Jun 2026 17:54:44 +0200 Subject: [PATCH 15/21] Fix QdrantStore.all_records missing with_vectors=True Qdrant's scroll() defaults to returning only payloads, not vectors. Without with_vectors=True, all 1963 records had empty embeddings, causing community detection to fail (0 communities) and topology to show insufficient_data. Now 1959/1963 records have 1024-dim embeddings and 1844 communities are detected. --- src/iai_mcp/qdrant_store.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/iai_mcp/qdrant_store.py b/src/iai_mcp/qdrant_store.py index ec6222d..1cc21da 100644 --- a/src/iai_mcp/qdrant_store.py +++ b/src/iai_mcp/qdrant_store.py @@ -644,6 +644,7 @@ class QdrantStore: offset=offset, scroll_filter=table_filter, with_payload=True, + with_vectors=True, ) all_points.extend(points) if next_offset is None: From 2cab42355f58dd0ce8bda84caf0d6f72acc26f00 Mon Sep 17 00:00:00 2001 From: Apunkt Date: Fri, 12 Jun 2026 18:07:24 +0200 Subject: [PATCH 16/21] Fix CLS clustering: include temporal_next edges in cluster building _build_hebbian_clusters() only used hebbian edges, but 97% of hebbian edges are self-loops from reinforce_record(). Only 7 are cross-record. temporal_next edges (140) provide the cross-record connectivity needed to form clusters >= CLUSTER_MIN_SIZE. Without them, consolidation never creates summaries, no consolidated_from edges exist, and the graph remains fragmented with 514 isolated nodes and max CC=58. --- src/iai_mcp/sleep.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/iai_mcp/sleep.py b/src/iai_mcp/sleep.py index e3e6c5b..f2d9b1c 100644 --- a/src/iai_mcp/sleep.py +++ b/src/iai_mcp/sleep.py @@ -296,16 +296,22 @@ def run_light_consolidation( def _build_hebbian_clusters(store: MemoryStore) -> list[list[UUID]]: - """Find connected components in the hebbian edge graph with size >= CLUSTER_MIN_SIZE.""" + """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. + """ edges_df = store.db.open_table(EDGES_TABLE).to_pandas() if edges_df.empty: return [] - hebbian = edges_df[edges_df["edge_type"] == "hebbian"] - if hebbian.empty: + relevant = edges_df[edges_df["edge_type"].isin(("hebbian", "temporal_next"))] + if relevant.empty: return [] adj: dict[UUID, set[UUID]] = {} - for _, row in hebbian.iterrows(): + for _, row in relevant.iterrows(): src = UUID(row["src"]) dst = UUID(row["dst"]) adj.setdefault(src, set()).add(dst) From bc3fc9d347e199239cb2fafa1ff6b3a912cda7a2 Mon Sep 17 00:00:00 2001 From: renovate-bot Date: Fri, 12 Jun 2026 16:27:03 +0000 Subject: [PATCH 17/21] chore(deps): update dependency @types/node to v25.9.3 --- mcp-wrapper/package-lock.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index 3d9bcdc..eca248f 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -517,9 +517,9 @@ } }, "node_modules/@types/node": { - "version": "25.9.1", - "resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.1.tgz", - "integrity": "sha512-xfrlY7UD5rMJk3ZVJP8BNzS28J36YJg+xp+LPXV1TdWxr8uMH5A860QNxYDGQe/ylDSgjxE52Q9VnO7p75tJxg==", + "version": "25.9.3", + "resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.3.tgz", + "integrity": "sha512-603BddQMv3pUcr4U2dhujk83N2tTDVr/34wII2B6bJy6g+8WD6yUb11jszNs0gdi4PesVWl7ABt8nYMVpnLUcg==", "dev": true, "license": "MIT", "dependencies": { From 581fb1ff47710969c6d6a0752e9935a73e5b7b53 Mon Sep 17 00:00:00 2001 From: renovate-bot Date: Fri, 12 Jun 2026 16:27:08 +0000 Subject: [PATCH 18/21] chore(deps): update dependency tsx to v4.22.4 --- mcp-wrapper/package-lock.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index 3d9bcdc..346ec6a 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -1584,9 +1584,9 @@ } }, "node_modules/tsx": { - "version": "4.22.3", - "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.22.3.tgz", - "integrity": "sha512-mdoNxBC/cSQObGGVQ5Bpn5i+yv7j68gk3Nfm3wFjcJg3Z0Mix9jzAFfP12prmm5eVGmDKtp0yyArrs0Q+8gZHg==", + "version": "4.22.4", + "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.22.4.tgz", + "integrity": "sha512-X8EX+XV4QR5xCsrgxaED954zTDfY8KqlDtskKEL0cHhyS/P8b4IFOvGDQpsC9Q1XnLq915wEfwwY/zzskCtmhg==", "dev": true, "license": "MIT", "dependencies": { From 9d929ba5803b5a9459d814f4f52fcec0e3333abc Mon Sep 17 00:00:00 2001 From: renovate-bot Date: Fri, 3 Jul 2026 14:28:43 +0000 Subject: [PATCH 19/21] chore(deps): update dependency tsx to v4.23.0 --- mcp-wrapper/package-lock.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index 39e46f9..ff808f4 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -1584,9 +1584,9 @@ } }, "node_modules/tsx": { - "version": "4.22.4", - "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.22.4.tgz", - "integrity": "sha512-X8EX+XV4QR5xCsrgxaED954zTDfY8KqlDtskKEL0cHhyS/P8b4IFOvGDQpsC9Q1XnLq915wEfwwY/zzskCtmhg==", + "version": "4.23.0", + "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.0.tgz", + "integrity": "sha512-eUdUIaCr963q2h5u3+QwvYp0+eqPvn+egeqZUm0hwERCqqx1E3kK5ehbGCvqSE5MQAULr67ww0cA3jKc3YkM1w==", "dev": true, "license": "MIT", "dependencies": { From 3d61c0837eb1fb1f4d8ee2624b3091655ee5023f Mon Sep 17 00:00:00 2001 From: renovate-bot Date: Wed, 8 Jul 2026 07:29:08 +0000 Subject: [PATCH 20/21] chore(deps): update dependency @types/node to v25.9.5 --- mcp-wrapper/package-lock.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/mcp-wrapper/package-lock.json b/mcp-wrapper/package-lock.json index 39e46f9..2be58ab 100644 --- a/mcp-wrapper/package-lock.json +++ b/mcp-wrapper/package-lock.json @@ -517,9 +517,9 @@ } }, "node_modules/@types/node": { - "version": "25.9.3", - "resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.3.tgz", - "integrity": "sha512-603BddQMv3pUcr4U2dhujk83N2tTDVr/34wII2B6bJy6g+8WD6yUb11jszNs0gdi4PesVWl7ABt8nYMVpnLUcg==", + "version": "25.9.5", + "resolved": "https://registry.npmjs.org/@types/node/-/node-25.9.5.tgz", + "integrity": "sha512-OScDchr2fwuUmWdf4kZ9h7PcJiYDVInhJizG/biAq3cAvqwYktuy/TYGGdZNMtNTFUP7rnb0NU4TUdm82kt4Rg==", "dev": true, "license": "MIT", "dependencies": { From 2d6de461a02f6a51c61bce04609fefc2380c46ad Mon Sep 17 00:00:00 2001 From: Apunkt Date: Tue, 14 Jul 2026 16:32:15 +0200 Subject: [PATCH 21/21] fix(graph): defer igraph rebuild to O(1) instead of O(m) per edge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MemoryGraph.add_edge() rebuilt the entire igraph after each edge write once backend switched (at 500 nodes), making edge addition O(m × (n+m)) — 49.81s for 5931 edges, causing 30s MCP timeout. Defer rebuild via _ig_stale flag; rebuild once on first read (centrality) instead of after every write. Edge addition now 0.77s. --- src/iai_mcp/graph.py | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/src/iai_mcp/graph.py b/src/iai_mcp/graph.py index 0bba61c..4becfb5 100644 --- a/src/iai_mcp/graph.py +++ b/src/iai_mcp/graph.py @@ -42,6 +42,7 @@ 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" @@ -79,10 +80,11 @@ class MemoryGraph: self._nx.add_edge( str(src), str(dst), weight=weight, edge_type=edge_type ) - 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() + 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 # ------------------------------------------------------ backend switching @@ -105,11 +107,17 @@ 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) + g = ig.Graph(n=len(nodes), edges=edges, directed=False) # type: ignore[unbound] 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 @@ -124,6 +132,7 @@ 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)