From ed43354f4c05730de9a08a8efcdf66e240c2c773 Mon Sep 17 00:00:00 2001 From: CaspianG <259116618+CaspianG@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:52:52 +0300 Subject: [PATCH] Add portable experience provider contracts --- .github/workflows/tests.yml | 40 ++ .gitignore | 1 + pyproject.toml | 5 + sdk/typescript/README.md | 15 + sdk/typescript/package-lock.json | 33 ++ sdk/typescript/package.json | 29 ++ sdk/typescript/src/index.ts | 187 ++++++++ sdk/typescript/test/client.test.mjs | 81 ++++ sdk/typescript/tsconfig.json | 22 + tests/test_experience_portability.py | 216 +++++++++ tests/test_official_provider_contracts.py | 117 +++++ tests/test_provider_experience_contracts.py | 355 +++++++++++++++ wavemind/__init__.py | 22 + wavemind/api.py | 209 ++++++++- wavemind/experience.py | 160 +++++++ wavemind/experience_portability.py | 481 ++++++++++++++++++++ wavemind/integrations/__init__.py | 18 +- wavemind/integrations/anthropic.py | 215 +++++++++ wavemind/integrations/langgraph.py | 85 ++++ wavemind/integrations/mcp_experience.py | 132 ++++++ wavemind/integrations/openai_agents.py | 194 ++++++++ 21 files changed, 2615 insertions(+), 2 deletions(-) create mode 100644 sdk/typescript/README.md create mode 100644 sdk/typescript/package-lock.json create mode 100644 sdk/typescript/package.json create mode 100644 sdk/typescript/src/index.ts create mode 100644 sdk/typescript/test/client.test.mjs create mode 100644 sdk/typescript/tsconfig.json create mode 100644 tests/test_experience_portability.py create mode 100644 tests/test_official_provider_contracts.py create mode 100644 tests/test_provider_experience_contracts.py create mode 100644 wavemind/experience_portability.py create mode 100644 wavemind/integrations/anthropic.py create mode 100644 wavemind/integrations/mcp_experience.py create mode 100644 wavemind/integrations/openai_agents.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 2a8726dd..cd5fbc5b 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -46,3 +46,43 @@ jobs: - name: Run tests run: pytest -q + + typescript-sdk: + runs-on: ubuntu-latest + + defaults: + run: + working-directory: sdk/typescript + + steps: + - uses: actions/checkout@v5 + + - uses: actions/setup-node@v4 + with: + node-version: "22" + cache: npm + cache-dependency-path: sdk/typescript/package-lock.json + + - name: Install SDK dependencies + run: npm ci + + - name: Build and test SDK + run: npm test + + provider-contracts: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v5 + + - uses: actions/setup-python@v6 + with: + python-version: "3.11" + + - name: Install official provider SDKs + run: | + python -m pip install --upgrade pip + python -m pip install -e ".[dev,providers]" + + - name: Run official provider contract tests + run: pytest -q tests/test_official_provider_contracts.py diff --git a/.gitignore b/.gitignore index ba018a28..56a0dfb4 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,7 @@ __pycache__/ .venv/ venv/ env/ +node_modules/ build/ dist/ diff --git a/pyproject.toml b/pyproject.toml index c2268f89..daf0880f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,6 +90,11 @@ bench = [ langchain = [ "langchain-classic>=1.0", ] +providers = [ + "anthropic>=0.120,<1", + "langgraph>=1.2,<2", + "openai-agents>=0.19,<1", +] production = [ "annoy>=1.17", "faiss-cpu>=1.8", diff --git a/sdk/typescript/README.md b/sdk/typescript/README.md new file mode 100644 index 00000000..1dc6cbeb --- /dev/null +++ b/sdk/typescript/README.md @@ -0,0 +1,15 @@ +# WaveMind TypeScript HTTP SDK + +```ts +import { WaveMindClient } from "@wavemind/http"; + +const memory = new WaveMindClient({ baseUrl: "http://localhost:8000" }); +await memory.remember({ text: "The deployment uses a canary.", namespace: "agent" }); +const packet = await memory.compileExperiencePacket({ + query: "How should I deploy?", + namespace: "agent", +}); +``` + +The client has no runtime dependencies and works with Node.js 18+ or modern +browsers that provide `fetch`. diff --git a/sdk/typescript/package-lock.json b/sdk/typescript/package-lock.json new file mode 100644 index 00000000..63fc7dab --- /dev/null +++ b/sdk/typescript/package-lock.json @@ -0,0 +1,33 @@ +{ + "name": "@wavemind/http", + "version": "0.1.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "@wavemind/http", + "version": "0.1.0", + "license": "MIT", + "devDependencies": { + "typescript": "5.9.3" + }, + "engines": { + "node": ">=18" + } + }, + "node_modules/typescript": { + "version": "5.9.3", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", + "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", + "dev": true, + "license": "Apache-2.0", + "bin": { + "tsc": "bin/tsc", + "tsserver": "bin/tsserver" + }, + "engines": { + "node": ">=14.17" + } + } + } +} diff --git a/sdk/typescript/package.json b/sdk/typescript/package.json new file mode 100644 index 00000000..25c0a24d --- /dev/null +++ b/sdk/typescript/package.json @@ -0,0 +1,29 @@ +{ + "name": "@wavemind/http", + "version": "0.1.0", + "description": "Typed TypeScript client for the WaveMind HTTP API", + "type": "module", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "files": [ + "dist", + "README.md" + ], + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "scripts": { + "build": "tsc -p tsconfig.json", + "test": "npm run build && node --test test/*.test.mjs" + }, + "engines": { + "node": ">=18" + }, + "license": "MIT", + "devDependencies": { + "typescript": "5.9.3" + } +} diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts new file mode 100644 index 00000000..8b127b4a --- /dev/null +++ b/sdk/typescript/src/index.ts @@ -0,0 +1,187 @@ +export interface WaveMindClientOptions { + baseUrl: string; + apiKey?: string; + fetch?: typeof globalThis.fetch; +} + +export interface RememberInput { + text: string; + namespace?: string; + tags?: string[]; + ttl_seconds?: number; + metadata?: Record; + priority?: number; +} + +export interface QueryInput { + text: string; + namespace?: string; + top_k?: number; + tags?: string[]; + min_score?: number; +} + +export interface QueryResult { + id: number; + text: string; + score: number; + vector_score: number; + field_score: number; + graph_score: number; + namespace: string; + tags: string[]; + metadata: Record; +} + +export interface ExperiencePacketInput { + query: string; + namespace?: string; + token_budget?: number; + top_k?: number; + domains?: string[]; + task_types?: string[]; + tools?: string[]; + include_canary?: boolean; +} + +export interface ExperiencePacketItem { + experience_id: string; + version: number; + kind: string; + title: string; + excerpt: string; + score: number; + signals: Record; + citation: string; + detail_ref: string; + provenance: Record; + estimated_tokens: number; + canary: boolean; +} + +export interface ExperiencePacket { + schema: "wavemind.experience_packet.v1"; + namespace: string; + query: string; + token_budget: number; + estimated_tokens: number; + items: ExperiencePacketItem[]; + omitted_count: number; + generated_at: number; + compiler_policy: Record; + citations: string[]; +} + +export interface TrajectoryInput { + payload: unknown; + provider?: "openai" | "anthropic" | "mcp" | "generic"; + namespace?: string; + trajectory_id?: string; + trust?: string; + status?: string; + confidence?: number; +} + +export class WaveMindHTTPError extends Error { + readonly status: number; + readonly body: unknown; + + constructor(status: number, body: unknown) { + super(`WaveMind HTTP request failed with status ${status}`); + this.name = "WaveMindHTTPError"; + this.status = status; + this.body = body; + } +} + +export class WaveMindClient { + readonly baseUrl: string; + readonly apiKey?: string; + private readonly fetchImpl: typeof globalThis.fetch; + + constructor(options: WaveMindClientOptions) { + const baseUrl = options.baseUrl.trim().replace(/\/+$/, ""); + if (!baseUrl) { + throw new Error("baseUrl must not be empty"); + } + this.baseUrl = baseUrl; + if (options.apiKey !== undefined) { + this.apiKey = options.apiKey; + } + this.fetchImpl = options.fetch ?? globalThis.fetch; + if (!this.fetchImpl) { + throw new Error("A fetch implementation is required"); + } + } + + remember(input: RememberInput): Promise<{ id: number }> { + return this.request("POST", "/remember", input); + } + + query(input: QueryInput): Promise<{ results: QueryResult[] }> { + return this.request("POST", "/query", input); + } + + compileExperiencePacket( + input: ExperiencePacketInput, + ): Promise { + return this.request("POST", "/experience/packet", input); + } + + getExperience( + experienceId: string, + namespace = "default", + ): Promise> { + const path = + `/experience/${encodeURIComponent(experienceId)}` + + `?namespace=${encodeURIComponent(namespace)}`; + return this.request("GET", path); + } + + ingestTrajectory( + input: TrajectoryInput, + ): Promise> { + return this.request("POST", "/experience/trajectories", input); + } + + exportExperienceBundle( + namespace?: string, + ): Promise> { + return this.request("POST", "/experience/export", { namespace }); + } + + importExperienceBundle( + bundle: Record, + ): Promise> { + return this.request("POST", "/experience/import", { bundle }); + } + + private async request( + method: string, + path: string, + body?: unknown, + ): Promise { + const headers: Record = { + accept: "application/json", + }; + if (body !== undefined) { + headers["content-type"] = "application/json"; + } + if (this.apiKey) { + headers.authorization = `Bearer ${this.apiKey}`; + } + const init: RequestInit = { method, headers }; + if (body !== undefined) { + init.body = JSON.stringify(body); + } + const response = await this.fetchImpl(`${this.baseUrl}${path}`, init); + const contentType = response.headers.get("content-type") ?? ""; + const payload = contentType.includes("application/json") + ? await response.json() + : await response.text(); + if (!response.ok) { + throw new WaveMindHTTPError(response.status, payload); + } + return payload as T; + } +} diff --git a/sdk/typescript/test/client.test.mjs b/sdk/typescript/test/client.test.mjs new file mode 100644 index 00000000..42313b5d --- /dev/null +++ b/sdk/typescript/test/client.test.mjs @@ -0,0 +1,81 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { WaveMindClient, WaveMindHTTPError } from "../dist/index.js"; + + +test("client sends typed memory and experience requests", async () => { + const requests = []; + const fetch = async (url, init) => { + requests.push({ url, init }); + const body = url.endsWith("/remember") + ? { id: 7 } + : { + schema: "wavemind.experience_packet.v1", + namespace: "agent", + query: "recover", + token_budget: 200, + estimated_tokens: 12, + items: [], + omitted_count: 0, + generated_at: 1, + compiler_policy: {}, + citations: [], + }; + return new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }; + const client = new WaveMindClient({ + baseUrl: "https://memory.example.test/", + apiKey: "secret", + fetch, + }); + + assert.deepEqual( + await client.remember({ text: "A durable fact", namespace: "agent" }), + { id: 7 }, + ); + const packet = await client.compileExperiencePacket({ + query: "recover", + namespace: "agent", + token_budget: 200, + }); + assert.equal(packet.schema, "wavemind.experience_packet.v1"); + assert.equal(requests[0].url, "https://memory.example.test/remember"); + assert.equal(requests[0].init.headers.authorization, "Bearer secret"); + assert.deepEqual(JSON.parse(requests[1].init.body), { + query: "recover", + namespace: "agent", + token_budget: 200, + }); +}); + + +test("client encodes experience IDs and exposes structured HTTP errors", async () => { + const fetch = async (url) => { + assert.equal( + url, + "https://memory.example.test/experience/id%2Fwith%20space?namespace=a%2Fb", + ); + return new Response(JSON.stringify({ detail: "Experience not found" }), { + status: 404, + headers: { "content-type": "application/json" }, + }); + }; + const client = new WaveMindClient({ + baseUrl: "https://memory.example.test", + fetch, + }); + + await assert.rejects( + () => client.getExperience("id/with space", "a/b"), + (error) => { + assert.ok(error instanceof WaveMindHTTPError); + assert.equal(error.status, 404); + assert.deepEqual(error.body, { detail: "Experience not found" }); + return true; + }, + ); +}); diff --git a/sdk/typescript/tsconfig.json b/sdk/typescript/tsconfig.json new file mode 100644 index 00000000..58bb3329 --- /dev/null +++ b/sdk/typescript/tsconfig.json @@ -0,0 +1,22 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "lib": [ + "ES2022", + "DOM" + ], + "declaration": true, + "declarationMap": true, + "sourceMap": true, + "strict": true, + "noUncheckedIndexedAccess": true, + "exactOptionalPropertyTypes": true, + "outDir": "dist", + "rootDir": "src" + }, + "include": [ + "src/**/*.ts" + ] +} diff --git a/tests/test_experience_portability.py b/tests/test_experience_portability.py new file mode 100644 index 00000000..66a6894b --- /dev/null +++ b/tests/test_experience_portability.py @@ -0,0 +1,216 @@ +from __future__ import annotations + +import copy +import json + +import pytest + +from wavemind.experience import ( + ExperienceKind, + ExperienceRecord, + ExperienceSource, + ExperienceStatus, + SQLiteExperienceStore, + TrustClass, + parse_tool_trajectory, +) +from wavemind.experience_portability import ( + PORTABLE_EXPERIENCE_SCHEMA, + export_experience_bundle, + import_anthropic_memory, + import_conversation_jsonl, + import_experience_bundle, + import_mem0_json, + load_experience_bundle, + write_experience_bundle, +) + + +def _record() -> ExperienceRecord: + return ExperienceRecord.create( + id="exp_portable", + kind=ExperienceKind.SUCCESSFUL_STRATEGY, + title="Retry transient HTTP errors", + content="Retry HTTP 503 twice with bounded exponential backoff.", + namespace="agent-a", + confidence=0.91, + trust=TrustClass.VERIFIED_OPERATOR, + status=ExperienceStatus.SHADOW, + source=ExperienceSource( + provider="test", + source_type="verified_run", + source_id="run-7", + ), + metadata={"service": "checkout"}, + ) + + +def test_portable_bundle_round_trip_has_exact_semantic_parity(tmp_path) -> None: + source = SQLiteExperienceStore(tmp_path / "source.db") + target = SQLiteExperienceStore(tmp_path / "target.db") + try: + record = source.put(_record()) + trajectory = parse_tool_trajectory( + { + "steps": [ + { + "id": "call-1", + "kind": "tool_call", + "name": "checkout", + "input": {"order": 7}, + }, + { + "id": "result-1", + "kind": "tool_result", + "name": "checkout", + "output": {"ok": True}, + "success": True, + "parent_id": "call-1", + }, + ] + }, + provider="generic", + namespace="agent-a", + trajectory_id="trajectory-portable", + ) + source.restore_trajectory(trajectory) + source.add_candidate_validation( + record.id, + evidence_id="evaluation-1", + successful=True, + score=0.96, + metadata={"suite": "held-out"}, + ) + + path = tmp_path / "portable.json" + written = write_experience_bundle(source, path, namespace="agent-a") + report = import_experience_bundle(target, path) + + assert written["schema"] == PORTABLE_EXPERIENCE_SCHEMA + assert report.exact + assert report.parity == 1.0 + assert report.record_count == 1 + assert report.trajectory_count == 1 + assert report.validation_count == 1 + assert target.get(record.id) == record + assert target.get_trajectory(trajectory.id) == trajectory + assert target.candidate_validations(experience_id=record.id)[0][ + "evidence_id" + ] == "evaluation-1" + + second = import_experience_bundle(target, path) + assert second.exact + assert second.inserted_records == 0 + assert second.inserted_trajectories == 0 + assert len(target.candidate_validations(experience_id=record.id)) == 1 + finally: + source.close() + target.close() + + +def test_portable_bundle_rejects_tampering_and_manifest_mismatch(tmp_path) -> None: + store = SQLiteExperienceStore(tmp_path / "source.db") + try: + store.put(_record()) + bundle = export_experience_bundle(store) + + tampered = copy.deepcopy(bundle) + tampered["records"][0]["content"] = "silently replaced" + with pytest.raises(ValueError, match="checksum mismatch"): + load_experience_bundle(tampered) + + manifest_tampered = copy.deepcopy(bundle) + manifest_tampered["manifest"]["record_count"] = 99 + content = { + key: value + for key, value in manifest_tampered.items() + if key != "content_sha256" + } + import hashlib + + manifest_tampered["content_sha256"] = hashlib.sha256( + json.dumps( + content, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + with pytest.raises(ValueError, match="manifest record_count mismatch"): + load_experience_bundle(manifest_tampered) + finally: + store.close() + + +def test_mem0_and_conversation_jsonl_imports_are_deduplicated(tmp_path) -> None: + store = SQLiteExperienceStore(tmp_path / "imports.db") + try: + mem0 = { + "results": [ + { + "id": "m-1", + "memory": "The customer prefers concise replies.", + "metadata": {"type": "preference", "confidence": 0.8}, + } + ] + } + first = import_mem0_json(store, mem0, namespace="customer-1") + second = import_mem0_json(store, mem0, namespace="customer-1") + assert first[0].id == second[0].id + assert first[0].kind == ExperienceKind.PREFERENCE + assert first[0].trust == TrustClass.IMPORTED + + path = tmp_path / "conversation.jsonl" + path.write_text( + json.dumps( + { + "id": "msg-1", + "role": "user", + "content": [{"type": "text", "text": "Never call after 18:00."}], + "preference": True, + } + ) + + "\n", + encoding="utf-8", + ) + imported = import_conversation_jsonl( + store, + path, + namespace="customer-1", + ) + assert imported[0].kind == ExperienceKind.PREFERENCE + assert imported[0].trust == TrustClass.EXPLICIT_USER + assert len(store.list(namespace="customer-1", include_expired=True)) == 2 + finally: + store.close() + + +def test_anthropic_memory_import_stays_inside_memory_root(tmp_path) -> None: + store = SQLiteExperienceStore(tmp_path / "anthropic.db") + try: + imported = import_anthropic_memory( + store, + { + "/memories/procedures/deploy.md": "Run the canary before production.", + "/memories/preferences.md": "Use concise release notes.", + }, + namespace="agent-a", + ) + assert {item.source.uri for item in imported} == { + "/memories/procedures/deploy.md", + "/memories/preferences.md", + } + assert all(item.status == ExperienceStatus.SHADOW for item in imported) + + invalid = ( + "/etc/passwd", + "/memories/../secret", + r"\memories\..\secret", + "/memories/%2e%2e/secret", + "/memories-safe/secret", + ) + for path in invalid: + with pytest.raises(ValueError, match="memory path|encoded traversal"): + import_anthropic_memory(store, {path: "blocked"}) + finally: + store.close() diff --git a/tests/test_official_provider_contracts.py b/tests/test_official_provider_contracts.py new file mode 100644 index 00000000..37360e95 --- /dev/null +++ b/tests/test_official_provider_contracts.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +import asyncio +from typing import TypedDict + +import pytest + +from wavemind import ( + ExperienceCompiler, + ExperienceKind, + ExperienceRecord, + ExperienceSource, + ExperienceStatus, + MemoryFirewall, + MemoryFirewallPolicy, + SQLiteExperienceStore, + TrustClass, +) +from wavemind.integrations.anthropic import ANTHROPIC_MEMORY_TOOL +from wavemind.integrations.langgraph import make_experience_recall_node +from wavemind.integrations.mcp_experience import build_experience_mcp_server +from wavemind.integrations.openai_agents import WaveMindAgentsSession + + +@pytest.fixture +def compiler(tmp_path): + store = SQLiteExperienceStore(tmp_path / "official-contracts.db") + store.put( + ExperienceRecord.create( + id="exp_official", + kind=ExperienceKind.PROCEDURE, + title="Official provider contract", + content="Retrieve the verified experience before acting.", + namespace="agent", + trust=TrustClass.VERIFIED_OPERATOR, + status=ExperienceStatus.ACTIVE, + source=ExperienceSource( + provider="test", + source_type="verified", + source_id="official-1", + ), + ) + ) + value = ExperienceCompiler( + store, + MemoryFirewall(MemoryFirewallPolicy(namespace="agent")), + ) + try: + yield value + finally: + store.close() + + +def test_openai_agents_runtime_session_protocol(tmp_path) -> None: + agents_memory = pytest.importorskip("agents.memory") + session = WaveMindAgentsSession( + "official-session", + db_path=tmp_path / "openai.db", + ) + try: + assert isinstance(session, agents_memory.Session) + asyncio.run(session.add_items([{"role": "user", "content": "hello"}])) + assert asyncio.run(session.get_items()) == [ + {"role": "user", "content": "hello"} + ] + finally: + session.close() + + +def test_anthropic_tool_definition_matches_official_typed_dict() -> None: + beta = pytest.importorskip("anthropic.types.beta") + annotations = beta.BetaMemoryTool20250818Param.__annotations__ + assert set(ANTHROPIC_MEMORY_TOOL) <= set(annotations) + assert ANTHROPIC_MEMORY_TOOL == { + "type": "memory_20250818", + "name": "memory", + } + + +def test_mcp_fastmcp_registers_experience_tools(compiler) -> None: + pytest.importorskip("mcp.server.fastmcp") + server = build_experience_mcp_server(compiler) + tools = asyncio.run(server.list_tools()) + assert {tool.name for tool in tools} == { + "compile_experience_packet", + "expand_experience", + } + assert all(tool.inputSchema["type"] == "object" for tool in tools) + + +def test_langgraph_compiles_and_invokes_experience_node(compiler) -> None: + graph_module = pytest.importorskip("langgraph.graph") + + class State(TypedDict, total=False): + input: str + experience_packet: str + experience_packet_data: dict + + builder = graph_module.StateGraph(State) + builder.add_node( + "experience", + make_experience_recall_node( + compiler, + namespace="agent", + token_budget=200, + ), + ) + builder.add_edge(graph_module.START, "experience") + builder.add_edge("experience", graph_module.END) + graph = builder.compile() + + result = graph.invoke({"input": "official provider contract"}) + assert "Official provider contract" in result["experience_packet"] + assert ( + result["experience_packet_data"]["items"][0]["experience_id"] + == "exp_official" + ) diff --git a/tests/test_provider_experience_contracts.py b/tests/test_provider_experience_contracts.py new file mode 100644 index 00000000..24b6cd5d --- /dev/null +++ b/tests/test_provider_experience_contracts.py @@ -0,0 +1,355 @@ +from __future__ import annotations + +import asyncio + +import pytest +from fastapi.testclient import TestClient + +from wavemind import ( + ExperienceCompiler, + ExperienceKind, + ExperienceRecord, + ExperienceSource, + ExperienceStatus, + MemoryFirewall, + MemoryFirewallPolicy, + SQLiteExperienceStore, + TrustClass, + WaveMind, +) +from wavemind.api import create_app +from wavemind.integrations.anthropic import ( + ANTHROPIC_MEMORY_TOOL, + AnthropicMemoryHandler, +) +from wavemind.integrations.langgraph import ( + make_experience_capture_node, + make_experience_recall_node, +) +from wavemind.integrations.mcp_experience import ExperienceMCPAdapter +from wavemind.integrations.openai_agents import ( + WaveMindAgentsSession, + make_experience_input_callback, +) + + +@pytest.fixture +def compiler(tmp_path): + store = SQLiteExperienceStore(tmp_path / "experience.db") + firewall = MemoryFirewall( + MemoryFirewallPolicy(namespace="agent", policy_id="provider-contract") + ) + value = ExperienceCompiler(store, firewall) + store.put( + ExperienceRecord.create( + id="exp_deploy", + kind=ExperienceKind.SUCCESSFUL_STRATEGY, + title="Recover a failed deployment", + content="Inspect health and logs, then roll back the failing release.", + namespace="agent", + confidence=0.95, + trust=TrustClass.VERIFIED_OPERATOR, + status=ExperienceStatus.ACTIVE, + source=ExperienceSource( + provider="evaluation", + source_type="verified_run", + source_id="deploy-7", + ), + ) + ) + try: + yield value + finally: + store.close() + + +def test_openai_agents_session_matches_async_session_contract(tmp_path) -> None: + db_path = tmp_path / "agents.db" + + async def exercise() -> None: + session = WaveMindAgentsSession("thread-1", db_path=db_path) + await session.add_items( + [ + {"role": "user", "content": "Deploy failed"}, + {"role": "assistant", "content": "Checking health"}, + ] + ) + assert [item["role"] for item in await session.get_items()] == [ + "user", + "assistant", + ] + assert await session.get_items(limit=1) == [ + {"role": "assistant", "content": "Checking health"} + ] + assert await session.pop_item() == { + "role": "assistant", + "content": "Checking health", + } + session.close() + + restarted = WaveMindAgentsSession("thread-1", db_path=db_path) + assert await restarted.get_items() == [ + {"role": "user", "content": "Deploy failed"} + ] + await restarted.clear_session() + assert await restarted.get_items() == [] + restarted.close() + + asyncio.run(exercise()) + + +def test_openai_input_callback_injects_ephemeral_packet_only(compiler) -> None: + callback = make_experience_input_callback( + compiler, + namespace="agent", + token_budget=200, + ) + + result = asyncio.run( + callback( + [{"role": "user", "content": "Earlier message"}], + [{"role": "user", "content": "How should I recover the deployment?"}], + ) + ) + + assert result[0]["content"] == "Earlier message" + assert result[-1]["content"] == "How should I recover the deployment?" + injected = result[-2] + assert injected["role"] == "system" + assert "Recover a failed deployment" in injected["content"] + assert injected["metadata"]["ephemeral"] is True + assert injected["metadata"]["citations"] == ["experience:exp_deploy@v1"] + + +def test_anthropic_memory_handler_implements_official_commands(tmp_path) -> None: + assert ANTHROPIC_MEMORY_TOOL == { + "type": "memory_20250818", + "name": "memory", + } + handler = AnthropicMemoryHandler( + str(tmp_path / "anthropic.db"), + namespace="agent", + ) + try: + assert handler.execute( + "create", + "/memories/deploy.md", + file_text="Check health.\nRoll back.", + )["created"] + assert handler.execute("view", "/memories")["files"] == [ + "/memories/deploy.md" + ] + handler.execute( + "str_replace", + "/memories/deploy.md", + old_str="Check health.", + new_str="Check health and logs.", + ) + handler.execute( + "insert", + "/memories/deploy.md", + insert_line=1, + insert_text="Confirm the failing revision.", + ) + viewed = handler.execute("view", "/memories/deploy.md") + assert viewed["content"].splitlines() == [ + "Check health and logs.", + "Confirm the failing revision.", + "Roll back.", + ] + handler.execute( + "rename", + "/memories/deploy.md", + new_path="/memories/recovery.md", + ) + assert list(handler.export_files()) == ["/memories/recovery.md"] + assert handler.execute("delete", "/memories/recovery.md")["deleted"] + + with pytest.raises(ValueError, match="memory path|encoded traversal"): + handler.execute("view", "/memories/../secrets") + finally: + handler.close() + + +def test_mcp_tools_list_and_call_follow_mcp_shapes(compiler) -> None: + adapter = ExperienceMCPAdapter(compiler) + tools = adapter.list_tools() + assert {tool["name"] for tool in tools} == { + "compile_experience_packet", + "expand_experience", + } + assert all(tool["inputSchema"]["type"] == "object" for tool in tools) + + packet = adapter.call_tool( + "compile_experience_packet", + { + "query": "recover deployment", + "namespace": "agent", + "token_budget": 200, + }, + ) + assert packet["schema"] == "wavemind.experience_packet.v1" + assert packet["items"][0]["experience_id"] == "exp_deploy" + + details = adapter.call_tool( + "expand_experience", + { + "experience_ids": ["exp_deploy"], + "namespace": "agent", + }, + ) + assert details["schema"] == "wavemind.experience_details.v1" + assert details["items"][0]["citation"] == "experience:exp_deploy@v1" + + +def test_langgraph_nodes_use_state_mapping_without_framework_dependency( + compiler, +) -> None: + recall = make_experience_recall_node( + compiler, + namespace="agent", + token_budget=200, + ) + state = recall({"input": "recover deployment"}) + assert "Recover a failed deployment" in state["experience_packet"] + assert state["experience_packet_data"]["items"][0]["experience_id"] == "exp_deploy" + + capture = make_experience_capture_node( + compiler, + namespace="agent", + kind=ExperienceKind.GOTCHA, + ) + captured = capture( + { + "thread_id": "thread-7", + "experience_title": "Deployment gotcha", + "experience": "The health endpoint remains cached for ten seconds.", + "success": True, + } + ) + stored = compiler.store.get(captured["wavemind_experience_id"]) + assert stored is not None + assert stored.kind == ExperienceKind.GOTCHA + assert stored.status == ExperienceStatus.SHADOW + + +def test_http_experience_contract_supports_packet_trajectory_and_bundle( + tmp_path, +) -> None: + experience_store = SQLiteExperienceStore(tmp_path / "http-experience.db") + experience_store.put( + ExperienceRecord.create( + id="exp_http", + kind=ExperienceKind.PROCEDURE, + title="HTTP recovery", + content="Inspect the failing health check before rollback.", + namespace="agent", + trust=TrustClass.VERIFIED_OPERATOR, + status=ExperienceStatus.ACTIVE, + source=ExperienceSource( + provider="test", + source_type="verified", + source_id="http-1", + ), + ) + ) + mind = WaveMind(db_path=tmp_path / "memory.db") + try: + with TestClient( + create_app(mind=mind, experience_store=experience_store) + ) as client: + packet = client.post( + "/experience/packet", + json={ + "query": "recover health check", + "namespace": "agent", + "token_budget": 200, + }, + ) + assert packet.status_code == 200 + assert packet.json()["items"][0]["experience_id"] == "exp_http" + + detail = client.get( + "/experience/exp_http", + params={"namespace": "agent"}, + ) + assert detail.status_code == 200 + assert detail.json()["citation"] == "experience:exp_http@v1" + + trajectory = client.post( + "/experience/trajectories", + json={ + "provider": "generic", + "namespace": "agent", + "trajectory_id": "http-trajectory", + "payload": { + "steps": [ + { + "id": "call-http", + "kind": "tool_call", + "name": "health", + }, + { + "id": "result-http", + "kind": "tool_result", + "name": "health", + "success": True, + "parent_id": "call-http", + }, + ] + }, + }, + ) + assert trajectory.status_code == 200 + assert trajectory.json()["trajectory"]["id"] == "http-trajectory" + assert trajectory.json()["inserted"] is True + replay = client.post( + "/experience/trajectories", + json={ + "provider": "generic", + "namespace": "agent", + "trajectory_id": "http-trajectory", + "payload": { + "steps": [ + { + "id": "call-http", + "kind": "tool_call", + "name": "health", + }, + { + "id": "result-http", + "kind": "tool_result", + "name": "health", + "success": True, + "parent_id": "call-http", + }, + ] + }, + }, + ) + assert replay.status_code == 200 + assert replay.json()["inserted"] is False + assert ( + replay.json()["experience"]["id"] + == trajectory.json()["experience"]["id"] + ) + + exported = client.post( + "/experience/export", + json={"namespace": "agent"}, + ) + assert exported.status_code == 200 + assert exported.json()["manifest"] == { + "record_count": 2, + "trajectory_count": 1, + "validation_count": 0, + } + imported = client.post( + "/experience/import", + json={"bundle": exported.json()}, + ) + assert imported.status_code == 200 + assert imported.json()["parity"] == 1.0 + finally: + mind.close() + experience_store.close() diff --git a/wavemind/__init__.py b/wavemind/__init__.py index 69d0326d..35675072 100644 --- a/wavemind/__init__.py +++ b/wavemind/__init__.py @@ -59,6 +59,18 @@ ExperiencePacket, ExperiencePacketItem, ) +from .experience_portability import ( + PORTABLE_EXPERIENCE_SCHEMA, + PortableImportReport, + experience_bundle_parity, + export_experience_bundle, + import_anthropic_memory, + import_conversation_jsonl, + import_experience_bundle, + import_mem0_json, + load_experience_bundle, + write_experience_bundle, +) from .memory_firewall import ( FirewallAction, FirewallContext, @@ -352,6 +364,8 @@ "ExperienceDetail", "ExperiencePacket", "ExperiencePacketItem", + "PORTABLE_EXPERIENCE_SCHEMA", + "PortableImportReport", "FirewallAction", "FirewallContext", "FirewallDecision", @@ -535,13 +549,20 @@ "default_memory_red_team_cases", "event_payload", "experience_from_trajectory", + "experience_bundle_parity", + "export_experience_bundle", "estimate_production_cost", "evaluate_production_slo", "evaluate_production_admission", "graph_payload", "image_payload", "ingest_jsonl_trajectories", + "import_anthropic_memory", + "import_conversation_jsonl", + "import_experience_bundle", + "import_mem0_json", "iter_jsonl_trajectories", + "load_experience_bundle", "kubernetes_resource_path", "operator_bundle", "operator_loop", @@ -585,5 +606,6 @@ "validate_postgres_pitr_commands", "video_payload", "write_memory_os_policy_evolution_artifacts", + "write_experience_bundle", "write_production_evidence_env_artifacts", ] diff --git a/wavemind/api.py b/wavemind/api.py index bd37880b..db426b1c 100644 --- a/wavemind/api.py +++ b/wavemind/api.py @@ -20,6 +20,18 @@ from .cluster import ClusterNode, build_cluster_autoscale_plan, build_cluster_plan from .core import WaveMind from .encoders import create_text_encoder +from .experience import ( + ExperienceStatus, + SQLiteExperienceStore, + TrustClass, + experience_from_trajectory, + parse_tool_trajectory, +) +from .experience_compiler import ExperienceCompiler +from .experience_portability import ( + export_experience_bundle, + import_experience_bundle, +) from .importers import import_path from .jobs import ( CachePrewarmWorker, @@ -35,6 +47,11 @@ query_with_vector_cache, ) from .observability import configure_observability, instrument_fastapi_app +from .memory_firewall import ( + FirewallContext, + MemoryFirewall, + MemoryFirewallPolicy, +) from .studio import STUDIO_HTML, field_heatmap, studio_snapshot @@ -1053,6 +1070,35 @@ class ConsolidateResponse(BaseModel): concepts: list[dict[str, Any]] +class ExperiencePacketRequest(BaseModel): + query: str = Field(min_length=1) + namespace: str = Field(default="default", min_length=1) + token_budget: int = Field(default=800, ge=32) + top_k: int = Field(default=8, ge=1, le=100) + domains: list[str] = Field(default_factory=list) + task_types: list[str] = Field(default_factory=list) + tools: list[str] = Field(default_factory=list) + include_canary: bool = False + + +class ExperienceTrajectoryRequest(BaseModel): + payload: Any + provider: str | None = None + namespace: str = Field(default="default", min_length=1) + trajectory_id: str | None = None + trust: str = TrustClass.TOOL_OUTPUT.value + status: str = ExperienceStatus.SHADOW.value + confidence: float = Field(default=0.5, ge=0.0, le=1.0) + + +class ExperienceBundleRequest(BaseModel): + namespace: str | None = None + + +class ExperienceBundleImportRequest(BaseModel): + bundle: dict[str, Any] + + def _remember_response_id(result: Any) -> int: if isinstance(result, int): return result @@ -1203,7 +1249,11 @@ def build_default_mind() -> WaveMind: ) -def create_app(mind: WaveMind | None = None) -> FastAPI: +def create_app( + mind: WaveMind | None = None, + *, + experience_store: SQLiteExperienceStore | None = None, +) -> FastAPI: logging.basicConfig(level=os.environ.get("WAVEMIND_LOG_LEVEL", "INFO")) app = FastAPI(title="WaveMind", version=__version__) observability = configure_observability(service_version=__version__) @@ -1226,6 +1276,52 @@ def create_app(mind: WaveMind | None = None) -> FastAPI: app.state.operation_metrics = APIOperationMetrics( max_samples=int(os.environ.get("WAVEMIND_METRICS_SAMPLE_SIZE", "512")) ) + app.state.experience_store = experience_store + app.state.experience_store_owned = False + app.state.experience_compilers = {} + app.state.experience_lock = Lock() + + def _experience_compiler(namespace: str) -> ExperienceCompiler: + selected = namespace.strip() + if not selected: + raise HTTPException(status_code=422, detail="namespace must not be empty") + with app.state.experience_lock: + store = app.state.experience_store + if store is None: + store = SQLiteExperienceStore( + Path( + os.environ.get( + "WAVEMIND_EXPERIENCE_DB", + "wavemind-experience.db", + ) + ) + ) + app.state.experience_store = store + app.state.experience_store_owned = True + compiler = app.state.experience_compilers.get(selected) + if compiler is None: + compiler = ExperienceCompiler( + store, + MemoryFirewall( + MemoryFirewallPolicy( + namespace=selected, + policy_id=f"http-api:{selected}", + ) + ), + ) + app.state.experience_compilers[selected] = compiler + return compiler + + def _close_experience_store() -> None: + if ( + app.state.experience_store_owned + and app.state.experience_store is not None + ): + app.state.experience_store.close() + app.state.experience_store = None + app.state.experience_compilers.clear() + + app.router.add_event_handler("shutdown", _close_experience_store) def _query_results(request: QueryRequest): if app.state.cache is None: @@ -1489,6 +1585,117 @@ def query(request: QueryRequest) -> QueryResponse: results = _query_results(request) return QueryResponse(results=_query_result_responses(results)) + @app.post( + "/experience/packet", + dependencies=[Depends(require_role("read"))], + ) + def compile_experience_packet(request: ExperiencePacketRequest): + compiler = _experience_compiler(request.namespace) + packet = compiler.compile_packet( + request.query, + namespace=request.namespace, + context=FirewallContext( + namespace=request.namespace, + actor="http_api", + ), + token_budget=request.token_budget, + top_k=request.top_k, + domains=request.domains, + task_types=request.task_types, + tools=request.tools, + include_canary=request.include_canary, + ) + return packet.as_dict() + + @app.get( + "/experience/{experience_id}", + dependencies=[Depends(require_role("read"))], + ) + def expand_experience(experience_id: str, namespace: str = "default"): + compiler = _experience_compiler(namespace) + details = compiler.expand( + [experience_id], + namespace=namespace, + context=FirewallContext(namespace=namespace, actor="http_api"), + ) + if not details: + raise HTTPException(status_code=404, detail="Experience not found") + return details[0].__dict__ + + @app.post( + "/experience/trajectories", + dependencies=[Depends(require_role("write"))], + ) + def ingest_experience_trajectory(request: ExperienceTrajectoryRequest): + try: + trajectory = parse_tool_trajectory( + request.payload, + provider=request.provider, + namespace=request.namespace, + trajectory_id=request.trajectory_id, + ) + record = experience_from_trajectory( + trajectory, + trust=TrustClass(request.trust), + status=ExperienceStatus(request.status), + confidence=request.confidence, + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + compiler = _experience_compiler(request.namespace) + with app.state.experience_lock: + existing = compiler.store.get_trajectory_by_source( + namespace=request.namespace, + source_sha256=trajectory.source_sha256, + ) + if existing is not None: + records = compiler.store.list_for_trajectory(existing.id) + return { + "experience": records[0].as_dict() if records else None, + "trajectory": existing.as_dict(), + "firewall": None, + "inserted": False, + } + stored, decision = compiler.submit( + record, + context=FirewallContext( + namespace=request.namespace, + actor="http_api", + actor_trust=TrustClass.TOOL_OUTPUT, + ), + ) + compiler.store.restore_trajectory(trajectory) + return { + "experience": stored.as_dict(), + "trajectory": trajectory.as_dict(), + "firewall": decision.as_dict(), + "inserted": True, + } + + @app.post( + "/experience/export", + dependencies=[Depends(require_role("admin"))], + ) + def export_experiences(request: ExperienceBundleRequest): + compiler = _experience_compiler(request.namespace or "default") + return export_experience_bundle( + compiler.store, + namespace=request.namespace, + ) + + @app.post( + "/experience/import", + dependencies=[Depends(require_role("admin"))], + ) + def import_experiences(request: ExperienceBundleImportRequest): + namespace = str(request.bundle.get("namespace") or "default") + compiler = _experience_compiler(namespace) + try: + report = import_experience_bundle(compiler.store, request.bundle) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + return report.__dict__ + @app.post( "/query/batch", response_model=QueryBatchResponse, diff --git a/wavemind/experience.py b/wavemind/experience.py index 11af9608..23978e81 100644 --- a/wavemind/experience.py +++ b/wavemind/experience.py @@ -1324,6 +1324,166 @@ def get_trajectory(self, trajectory_id: str) -> ToolTrajectory | None: steps=tuple(_trajectory_step_from_row(item) for item in step_rows), ) + def get_trajectory_by_source( + self, + *, + namespace: str, + source_sha256: str, + ) -> ToolTrajectory | None: + with self._lock: + row = self.conn.execute( + """ + SELECT id FROM experience_trajectories + WHERE namespace = ? AND source_sha256 = ? + """, + (namespace, source_sha256), + ).fetchone() + return self.get_trajectory(str(row["id"])) if row is not None else None + + def list_for_trajectory(self, trajectory_id: str) -> list[ExperienceRecord]: + with self._lock: + rows = self.conn.execute( + """ + SELECT * FROM experience_records + WHERE trajectory_id = ? + ORDER BY created_at, id + """, + (trajectory_id,), + ).fetchall() + return [_experience_from_row(row) for row in rows] + + def list_trajectories( + self, + *, + namespace: str | None = None, + limit: int = 10_000, + ) -> list[ToolTrajectory]: + if not 1 <= int(limit) <= 100_000: + raise ValueError("limit must be between 1 and 100000") + query = "SELECT id FROM experience_trajectories" + values: list[Any] = [] + if namespace is not None: + query += " WHERE namespace = ?" + values.append(namespace) + query += " ORDER BY created_at, id LIMIT ?" + values.append(int(limit)) + with self._lock: + rows = self.conn.execute(query, values).fetchall() + return [ + trajectory + for row in rows + if (trajectory := self.get_trajectory(str(row["id"]))) is not None + ] + + def restore_trajectory(self, trajectory: ToolTrajectory) -> bool: + with self._lock, self.conn: + existing = self.conn.execute( + "SELECT source_sha256 FROM experience_trajectories WHERE id = ?", + (trajectory.id,), + ).fetchone() + if existing is not None: + if str(existing["source_sha256"]) != trajectory.source_sha256: + raise ValueError( + f"trajectory id {trajectory.id!r} already exists with " + "different data" + ) + restored = self.get_trajectory(trajectory.id) + if restored != trajectory: + raise ValueError( + f"trajectory id {trajectory.id!r} failed exact replay validation" + ) + return False + digest_row = self.conn.execute( + """ + SELECT id FROM experience_trajectories + WHERE namespace = ? AND source_sha256 = ? + """, + (trajectory.namespace, trajectory.source_sha256), + ).fetchone() + if digest_row is not None: + raise ValueError( + "trajectory source already exists under a different id: " + f"{digest_row['id']}" + ) + self.conn.execute( + """ + INSERT INTO experience_trajectories ( + id, namespace, provider, source_sha256, started_at, ended_at, + metadata_json, raw_event_count, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + trajectory.id, + trajectory.namespace, + trajectory.provider, + trajectory.source_sha256, + trajectory.started_at, + trajectory.ended_at, + _json_dumps(trajectory.metadata), + int(trajectory.raw_event_count), + _now(), + ), + ) + for step in trajectory.steps: + self.conn.execute( + """ + INSERT INTO experience_trajectory_steps ( + trajectory_id, step_id, sequence, kind, name, input_json, + output_json, success, started_at, finished_at, parent_id, + metadata_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + trajectory.id, + step.id, + int(step.sequence), + step.kind.value, + step.name, + _json_dumps(step.input), + _json_dumps(step.output), + None if step.success is None else int(step.success), + step.started_at, + step.finished_at, + step.parent_id, + _json_dumps(step.metadata), + ), + ) + self._audit( + "trajectory_restored", + trajectory_id=trajectory.id, + metadata={"source_sha256": trajectory.source_sha256}, + ) + return True + + def candidate_validations( + self, + *, + experience_id: str | None = None, + ) -> list[dict[str, Any]]: + query = """ + SELECT experience_id, evidence_id, successful, score, + created_at, metadata_json + FROM experience_candidate_validations + """ + values: list[Any] = [] + if experience_id is not None: + query += " WHERE experience_id = ?" + values.append(experience_id) + query += " ORDER BY experience_id, created_at, id" + with self._lock: + rows = self.conn.execute(query, values).fetchall() + return [ + { + "experience_id": str(row["experience_id"]), + "evidence_id": str(row["evidence_id"]), + "successful": bool(row["successful"]), + "score": row["score"], + "created_at": float(row["created_at"]), + "metadata": _json_loads(row["metadata_json"], {}), + } + for row in rows + ] + def audit_events( self, *, diff --git a/wavemind/experience_portability.py b/wavemind/experience_portability.py new file mode 100644 index 00000000..b4bc0652 --- /dev/null +++ b/wavemind/experience_portability.py @@ -0,0 +1,481 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any, Mapping, Sequence + +from .experience import ( + ExperienceKind, + ExperienceRecord, + ExperienceSource, + ExperienceStatus, + SQLiteExperienceStore, + ToolTrajectory, + TrajectoryStep, + TrustClass, +) + + +PORTABLE_EXPERIENCE_SCHEMA = "wavemind.portable_experience.v1" + + +@dataclass(frozen=True) +class PortableImportReport: + record_count: int + trajectory_count: int + validation_count: int + inserted_records: int + inserted_trajectories: int + parity: float + source_sha256: str + + @property + def exact(self) -> bool: + return self.parity == 1.0 + + +def export_experience_bundle( + store: SQLiteExperienceStore, + *, + namespace: str | None = None, +) -> dict[str, Any]: + records = store.list( + namespace=namespace, + include_expired=True, + limit=10_000, + ) + trajectories = store.list_trajectories(namespace=namespace, limit=100_000) + validations = [ + row + for row in store.candidate_validations() + if namespace is None + or ( + (record := store.get(str(row["experience_id"]))) is not None + and record.namespace == namespace + ) + ] + payload = { + "schema": PORTABLE_EXPERIENCE_SCHEMA, + "namespace": namespace, + "records": [record.as_dict() for record in records], + "trajectories": [trajectory.as_dict() for trajectory in trajectories], + "validations": validations, + "manifest": { + "record_count": len(records), + "trajectory_count": len(trajectories), + "validation_count": len(validations), + }, + } + payload["content_sha256"] = _bundle_sha256(payload) + return payload + + +def write_experience_bundle( + store: SQLiteExperienceStore, + path: str | Path, + *, + namespace: str | None = None, +) -> dict[str, Any]: + payload = export_experience_bundle(store, namespace=namespace) + destination = Path(path) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text( + json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n", + encoding="utf-8", + ) + return payload + + +def load_experience_bundle( + source: str | Path | Mapping[str, Any], +) -> dict[str, Any]: + if isinstance(source, Mapping): + payload = dict(source) + else: + payload = json.loads(Path(source).read_text(encoding="utf-8")) + if payload.get("schema") != PORTABLE_EXPERIENCE_SCHEMA: + raise ValueError("unsupported portable experience bundle schema") + expected = str(payload.get("content_sha256") or "") + if len(expected) != 64 or _bundle_sha256(payload) != expected: + raise ValueError("portable experience bundle checksum mismatch") + manifest = payload.get("manifest") or {} + for key, collection in ( + ("record_count", payload.get("records") or []), + ("trajectory_count", payload.get("trajectories") or []), + ("validation_count", payload.get("validations") or []), + ): + if int(manifest.get(key, -1)) != len(collection): + raise ValueError(f"portable experience manifest {key} mismatch") + return payload + + +def import_experience_bundle( + store: SQLiteExperienceStore, + source: str | Path | Mapping[str, Any], +) -> PortableImportReport: + payload = load_experience_bundle(source) + inserted_records = 0 + inserted_trajectories = 0 + for row in payload.get("records") or []: + record = ExperienceRecord.from_dict(row) + existing = store.get(record.id) + stored = store.put(record) + inserted_records += int(existing is None) + if stored.as_dict() != record.as_dict(): + raise ValueError(f"experience replay mismatch for {record.id}") + for row in payload.get("trajectories") or []: + trajectory = _trajectory_from_dict(row) + inserted_trajectories += int(store.restore_trajectory(trajectory)) + for row in payload.get("validations") or []: + store.add_candidate_validation( + str(row["experience_id"]), + evidence_id=str(row["evidence_id"]), + successful=bool(row["successful"]), + score=row.get("score"), + metadata=dict(row.get("metadata") or {}), + ) + replayed = export_experience_bundle( + store, + namespace=payload.get("namespace"), + ) + parity = experience_bundle_parity(payload, replayed) + return PortableImportReport( + record_count=len(payload.get("records") or []), + trajectory_count=len(payload.get("trajectories") or []), + validation_count=len(payload.get("validations") or []), + inserted_records=inserted_records, + inserted_trajectories=inserted_trajectories, + parity=parity, + source_sha256=str(payload["content_sha256"]), + ) + + +def experience_bundle_parity( + source: Mapping[str, Any], + target: Mapping[str, Any], +) -> float: + source_keys = _semantic_keys(source) + target_keys = _semantic_keys(target) + if not source_keys and not target_keys: + return 1.0 + union = source_keys | target_keys + return len(source_keys & target_keys) / len(union) + + +def import_mem0_json( + store: SQLiteExperienceStore, + source: str | Path | Mapping[str, Any] | Sequence[Mapping[str, Any]], + *, + namespace: str = "default", +) -> list[ExperienceRecord]: + payload = _load_json_source(source) + if isinstance(payload, Mapping): + rows = payload.get("results") or payload.get("memories") or payload.get("data") + if rows is None and ("memory" in payload or "text" in payload): + rows = [payload] + else: + rows = payload + if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes, bytearray)): + raise ValueError("Mem0 import requires a memory array") + imported = [] + for index, row in enumerate(rows): + if not isinstance(row, Mapping): + raise ValueError(f"Mem0 memory at index {index} must be an object") + content = str(row.get("memory") or row.get("text") or "").strip() + if not content: + continue + source_id = str(row.get("id") or row.get("memory_id") or f"mem0-{index}") + metadata = dict(row.get("metadata") or {}) + record = ExperienceRecord.create( + id=_import_id("mem0", namespace, source_id), + kind=_kind_from_metadata(metadata), + title=str(metadata.get("title") or f"Imported Mem0 memory {index + 1}"), + content=content, + source=ExperienceSource( + provider="mem0", + source_type="memory_export", + source_id=source_id, + metadata={"original_metadata": metadata}, + ), + namespace=namespace, + confidence=float(row.get("confidence") or metadata.get("confidence") or 0.5), + trust=TrustClass.IMPORTED, + status=ExperienceStatus.SHADOW, + metadata={"import_format": "mem0", **metadata}, + ) + imported.append( + _put_imported( + store, + record, + dedupe_key=f"mem0:{namespace}:{source_id}", + ) + ) + return imported + + +def import_conversation_jsonl( + store: SQLiteExperienceStore, + path: str | Path, + *, + namespace: str = "default", + max_line_bytes: int = 2 * 1024 * 1024, +) -> list[ExperienceRecord]: + imported = [] + with Path(path).open("rb") as handle: + for line_number, raw in enumerate(handle, start=1): + if len(raw) > max_line_bytes: + raise ValueError(f"conversation line {line_number} exceeds size limit") + if not raw.strip(): + continue + row = json.loads(raw) + if not isinstance(row, Mapping): + raise ValueError(f"conversation line {line_number} must be an object") + content = _conversation_content(row.get("content")) + if not content: + continue + role = str(row.get("role") or "unknown") + source_id = str( + row.get("id") + or row.get("message_id") + or f"{Path(path).name}:{line_number}" + ) + record = ExperienceRecord.create( + id=_import_id("chat", namespace, source_id), + kind=( + ExperienceKind.PREFERENCE + if role == "user" and row.get("preference") is True + else ExperienceKind.EPISODE + ), + title=f"Imported {role} message", + content=content, + source=ExperienceSource( + provider=str(row.get("provider") or "conversation_jsonl"), + source_type="conversation_message", + source_id=source_id, + metadata={"role": role}, + ), + namespace=namespace, + confidence=0.5, + trust=( + TrustClass.EXPLICIT_USER + if role == "user" + else TrustClass.IMPORTED + ), + status=ExperienceStatus.SHADOW, + metadata={"import_format": "conversation_jsonl", "role": role}, + ) + imported.append( + _put_imported( + store, + record, + dedupe_key=f"chat:{namespace}:{source_id}", + ) + ) + return imported + + +def import_anthropic_memory( + store: SQLiteExperienceStore, + source: str | Path | Mapping[str, str], + *, + namespace: str = "default", +) -> list[ExperienceRecord]: + files: dict[str, str] = {} + if isinstance(source, Mapping): + files = {str(path): str(content) for path, content in source.items()} + else: + root = Path(source) + if root.is_file(): + files[f"/memories/{root.name}"] = root.read_text(encoding="utf-8") + else: + for path in sorted(root.rglob("*")): + if path.is_file(): + relative = path.relative_to(root).as_posix() + files[f"/memories/{relative}"] = path.read_text(encoding="utf-8") + imported = [] + for path, content in files.items(): + normalized = validate_anthropic_memory_path(path) + if normalized == "/memories": + continue + source_id = normalized + record = ExperienceRecord.create( + id=_import_id("anthropic", namespace, source_id), + kind=ExperienceKind.PROCEDURE, + title=PurePosixPath(normalized).name, + content=content, + source=ExperienceSource( + provider="anthropic", + source_type="memory_tool_file", + source_id=source_id, + uri=normalized, + ), + namespace=namespace, + confidence=0.5, + trust=TrustClass.IMPORTED, + status=ExperienceStatus.SHADOW, + metadata={"import_format": "anthropic_memory", "path": normalized}, + ) + imported.append( + _put_imported( + store, + record, + dedupe_key=f"anthropic:{namespace}:{normalized}", + ) + ) + return imported + + +def _semantic_keys(payload: Mapping[str, Any]) -> set[str]: + keys = { + "record:" + hashlib.sha256( + json.dumps( + row, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + for row in payload.get("records") or [] + } + keys.update( + f"trajectory:{row.get('id')}:{row.get('source_sha256')}" + for row in payload.get("trajectories") or [] + ) + keys.update( + "validation:" + + ":".join( + ( + str(row.get("experience_id")), + str(row.get("evidence_id")), + str(bool(row.get("successful"))), + str(row.get("score")), + hashlib.sha256( + json.dumps( + row.get("metadata") or {}, + sort_keys=True, + separators=(",", ":"), + ).encode() + ).hexdigest(), + ) + ) + for row in payload.get("validations") or [] + ) + return keys + + +def _bundle_sha256(payload: Mapping[str, Any]) -> str: + content = {key: value for key, value in payload.items() if key != "content_sha256"} + return hashlib.sha256( + json.dumps( + content, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + + +def _import_id(provider: str, namespace: str, source_id: str) -> str: + digest = hashlib.sha256( + f"{provider}\0{namespace}\0{source_id}".encode("utf-8") + ).hexdigest() + return f"exp_{provider}_{digest[:24]}" + + +def _put_imported( + store: SQLiteExperienceStore, + record: ExperienceRecord, + *, + dedupe_key: str, +) -> ExperienceRecord: + existing = store.get(record.id) + if existing is None: + return store.put(record, dedupe_key=dedupe_key) + comparable_fields = ( + "id", + "namespace", + "kind", + "title", + "content", + "applicability", + "outcome", + "confidence", + "trust", + "source", + "trajectory", + "expires_at", + "status", + "supersedes_id", + "rollback_of_id", + "metadata", + ) + if any(getattr(existing, name) != getattr(record, name) for name in comparable_fields): + raise ValueError( + f"import source for experience {record.id!r} changed since first import" + ) + return existing + + +def _trajectory_from_dict(value: Mapping[str, Any]) -> ToolTrajectory: + return ToolTrajectory( + id=str(value["id"]), + provider=str(value["provider"]), + namespace=str(value.get("namespace") or "default"), + steps=tuple( + TrajectoryStep.from_dict(row, sequence=index) + for index, row in enumerate(value.get("steps") or []) + ), + source_sha256=str(value["source_sha256"]), + started_at=value.get("started_at"), + ended_at=value.get("ended_at"), + metadata=dict(value.get("metadata") or {}), + raw_event_count=int(value.get("raw_event_count") or 0), + ) + + +def _load_json_source( + source: str | Path | Mapping[str, Any] | Sequence[Mapping[str, Any]], +) -> Any: + if isinstance(source, (str, Path)): + return json.loads(Path(source).read_text(encoding="utf-8")) + return source + + +def _kind_from_metadata(metadata: Mapping[str, Any]) -> ExperienceKind: + value = str(metadata.get("type") or metadata.get("kind") or "fact") + try: + return ExperienceKind(value) + except ValueError: + return ExperienceKind.FACT + + +def _conversation_content(value: Any) -> str: + if isinstance(value, str): + return value.strip() + if isinstance(value, Sequence): + parts = [] + for item in value: + if isinstance(item, Mapping): + text = item.get("text") or item.get("content") + if text: + parts.append(str(text)) + elif item: + parts.append(str(item)) + return "\n".join(parts).strip() + return "" + + +def validate_anthropic_memory_path(value: str) -> str: + raw = str(value).replace("\\", "/") + lowered = raw.lower() + if "%2e" in lowered or "%2f" in lowered or "%5c" in lowered: + raise ValueError("Anthropic memory path contains encoded traversal") + path = PurePosixPath(raw) + if not raw.startswith("/memories") or ".." in path.parts: + raise ValueError("Anthropic memory paths must stay under /memories") + normalized = "/" + "/".join(part for part in path.parts if part != "/") + if normalized != "/memories" and not normalized.startswith("/memories/"): + raise ValueError("Anthropic memory paths must stay under /memories") + return normalized diff --git a/wavemind/integrations/__init__.py b/wavemind/integrations/__init__.py index 4909a0a9..24052400 100644 --- a/wavemind/integrations/__init__.py +++ b/wavemind/integrations/__init__.py @@ -1,15 +1,31 @@ """Optional integrations for external agent frameworks.""" +from .anthropic import ANTHROPIC_MEMORY_TOOL, AnthropicMemoryHandler from .autogen import WaveMindAutoGenMemory from .crewai import WaveMindCrewAITools -from .langgraph import make_persist_node, make_recall_node +from .langgraph import ( + make_experience_capture_node, + make_experience_recall_node, + make_persist_node, + make_recall_node, +) from .llamaindex import WaveMindNode, WaveMindRetriever +from .mcp_experience import ExperienceMCPAdapter, build_experience_mcp_server +from .openai_agents import WaveMindAgentsSession, make_experience_input_callback __all__ = [ + "ANTHROPIC_MEMORY_TOOL", + "AnthropicMemoryHandler", + "ExperienceMCPAdapter", "WaveMindAutoGenMemory", + "WaveMindAgentsSession", "WaveMindCrewAITools", "WaveMindNode", "WaveMindRetriever", + "build_experience_mcp_server", + "make_experience_capture_node", + "make_experience_input_callback", + "make_experience_recall_node", "make_persist_node", "make_recall_node", ] diff --git a/wavemind/integrations/anthropic.py b/wavemind/integrations/anthropic.py new file mode 100644 index 00000000..4f288da4 --- /dev/null +++ b/wavemind/integrations/anthropic.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import sqlite3 +from pathlib import PurePosixPath +from threading import RLock +from typing import Any + +from wavemind.experience_portability import validate_anthropic_memory_path + + +ANTHROPIC_MEMORY_TOOL = { + "type": "memory_20250818", + "name": "memory", +} + + +class AnthropicMemoryHandler: + """Client-side storage handler for Anthropic's official memory tool.""" + + tool_definition = ANTHROPIC_MEMORY_TOOL + + def __init__( + self, + db_path: str = "wavemind-anthropic.db", + *, + namespace: str = "default", + ): + self.namespace = str(namespace).strip() + if not self.namespace: + raise ValueError("namespace must not be empty") + self._lock = RLock() + self._conn = sqlite3.connect(db_path, check_same_thread=False, timeout=30.0) + self._conn.row_factory = sqlite3.Row + with self._conn: + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS anthropic_memory_files ( + namespace TEXT NOT NULL, + path TEXT NOT NULL, + content TEXT NOT NULL, + PRIMARY KEY (namespace, path) + ) + """ + ) + + def execute(self, command: str, path: str, **arguments: Any) -> dict[str, Any]: + selected = str(command).strip() + normalized = validate_anthropic_memory_path(path) + handlers = { + "view": self._view, + "create": self._create, + "str_replace": self._str_replace, + "insert": self._insert, + "delete": self._delete, + "rename": self._rename, + } + if selected not in handlers: + raise ValueError(f"unsupported Anthropic memory command: {selected}") + return handlers[selected](normalized, **arguments) + + def _view(self, path: str, **arguments: Any) -> dict[str, Any]: + with self._lock: + if path == "/memories" or arguments.get("directory"): + prefix = path.rstrip("/") + "/" + rows = self._conn.execute( + """ + SELECT path FROM anthropic_memory_files + WHERE namespace = ? AND path LIKE ? ORDER BY path + """, + (self.namespace, prefix + "%"), + ).fetchall() + return {"path": path, "files": [str(row["path"]) for row in rows]} + row = self._conn.execute( + """ + SELECT content FROM anthropic_memory_files + WHERE namespace = ? AND path = ? + """, + (self.namespace, path), + ).fetchone() + if row is None: + raise FileNotFoundError(path) + content = str(row["content"]) + view_range = arguments.get("view_range") + if view_range is not None: + start, end = (int(value) for value in view_range) + if start < 1 or end < start: + raise ValueError("view_range must contain positive ordered lines") + content = "\n".join(content.splitlines()[start - 1 : end]) + return {"path": path, "content": content} + + def _create(self, path: str, **arguments: Any) -> dict[str, Any]: + if path == "/memories": + raise ValueError("create requires a file path") + content = str(arguments.get("file_text", arguments.get("content", ""))) + with self._lock, self._conn: + existing = self._conn.execute( + """ + SELECT 1 FROM anthropic_memory_files + WHERE namespace = ? AND path = ? + """, + (self.namespace, path), + ).fetchone() + if existing is not None: + raise FileExistsError(path) + self._conn.execute( + """ + INSERT INTO anthropic_memory_files (namespace, path, content) + VALUES (?, ?, ?) + """, + (self.namespace, path, content), + ) + return {"path": path, "created": True} + + def _str_replace(self, path: str, **arguments: Any) -> dict[str, Any]: + old = str(arguments.get("old_str", "")) + new = str(arguments.get("new_str", "")) + if not old: + raise ValueError("old_str must not be empty") + content = self._content(path) + count = content.count(old) + if count != 1: + raise ValueError("old_str must match exactly once") + self._write(path, content.replace(old, new, 1)) + return {"path": path, "replaced": True} + + def _insert(self, path: str, **arguments: Any) -> dict[str, Any]: + line = int(arguments.get("insert_line", 0)) + text = str(arguments.get("insert_text", "")) + lines = self._content(path).splitlines() + if line < 0 or line > len(lines): + raise ValueError("insert_line is outside the file") + lines[line:line] = text.splitlines() + self._write(path, "\n".join(lines)) + return {"path": path, "inserted": True} + + def _delete(self, path: str, **_: Any) -> dict[str, Any]: + if path == "/memories": + raise ValueError("cannot delete the memory root") + with self._lock, self._conn: + cursor = self._conn.execute( + """ + DELETE FROM anthropic_memory_files + WHERE namespace = ? AND path = ? + """, + (self.namespace, path), + ) + if cursor.rowcount != 1: + raise FileNotFoundError(path) + return {"path": path, "deleted": True} + + def _rename(self, path: str, **arguments: Any) -> dict[str, Any]: + new_path = validate_anthropic_memory_path(str(arguments.get("new_path", ""))) + if new_path == "/memories": + raise ValueError("rename requires a file destination") + content = self._content(path) + with self._lock, self._conn: + conflict = self._conn.execute( + """ + SELECT 1 FROM anthropic_memory_files + WHERE namespace = ? AND path = ? + """, + (self.namespace, new_path), + ).fetchone() + if conflict is not None: + raise FileExistsError(new_path) + self._conn.execute( + """ + INSERT INTO anthropic_memory_files (namespace, path, content) + VALUES (?, ?, ?) + """, + (self.namespace, new_path, content), + ) + self._conn.execute( + """ + DELETE FROM anthropic_memory_files + WHERE namespace = ? AND path = ? + """, + (self.namespace, path), + ) + return {"path": path, "new_path": new_path, "renamed": True} + + def _content(self, path: str) -> str: + return str(self._view(path)["content"]) + + def _write(self, path: str, content: str) -> None: + with self._lock, self._conn: + cursor = self._conn.execute( + """ + UPDATE anthropic_memory_files SET content = ? + WHERE namespace = ? AND path = ? + """, + (content, self.namespace, path), + ) + if cursor.rowcount != 1: + raise FileNotFoundError(path) + + def export_files(self) -> dict[str, str]: + with self._lock: + rows = self._conn.execute( + """ + SELECT path, content FROM anthropic_memory_files + WHERE namespace = ? ORDER BY path + """, + (self.namespace,), + ).fetchall() + return {str(row["path"]): str(row["content"]) for row in rows} + + def close(self) -> None: + with self._lock: + self._conn.close() + + +def anthropic_memory_filename(path: str) -> str: + return PurePosixPath(validate_anthropic_memory_path(path)).name diff --git a/wavemind/integrations/langgraph.py b/wavemind/integrations/langgraph.py index ee9e7ca6..7f288e22 100644 --- a/wavemind/integrations/langgraph.py +++ b/wavemind/integrations/langgraph.py @@ -4,6 +4,16 @@ from typing import Any from wavemind.core import WaveMind +from wavemind.experience import ( + ExperienceKind, + ExperienceOutcome, + ExperienceRecord, + ExperienceSource, + ExperienceStatus, + TrustClass, +) +from wavemind.experience_compiler import ExperienceCompiler +from wavemind.memory_firewall import FirewallContext def make_recall_node( @@ -56,3 +66,78 @@ def persist_node(state: Mapping[str, Any]) -> dict[str, Any]: return {"wavemind_memory_ids": ids} return persist_node + + +def make_experience_recall_node( + compiler: ExperienceCompiler, + *, + namespace: str, + input_key: str = "input", + output_key: str = "experience_packet", + token_budget: int = 800, + top_k: int = 8, +) -> Callable[[Mapping[str, Any]], dict[str, Any]]: + """Create a LangGraph node that compiles trusted experience into state.""" + + def recall_node(state: Mapping[str, Any]) -> dict[str, Any]: + query = str(state.get(input_key, "")).strip() + if not query: + return {output_key: "", f"{output_key}_data": None} + packet = compiler.compile_packet( + query, + namespace=namespace, + context=FirewallContext(namespace=namespace, actor="langgraph"), + token_budget=token_budget, + top_k=top_k, + ) + return { + output_key: packet.as_prompt() if packet.items else "", + f"{output_key}_data": packet.as_dict(), + } + + return recall_node + + +def make_experience_capture_node( + compiler: ExperienceCompiler, + *, + namespace: str, + content_key: str = "experience", + title_key: str = "experience_title", + kind: ExperienceKind = ExperienceKind.EPISODE, +) -> Callable[[Mapping[str, Any]], dict[str, Any]]: + """Capture graph output as a shadow candidate guarded by the firewall.""" + + def capture_node(state: Mapping[str, Any]) -> dict[str, Any]: + content = str(state.get(content_key, "")).strip() + if not content: + return {"wavemind_experience_id": None} + title = str(state.get(title_key) or "LangGraph experience").strip() + record = ExperienceRecord.create( + kind=kind, + title=title, + content=content, + namespace=namespace, + outcome=ExperienceOutcome( + success=state.get("success") + if isinstance(state.get("success"), bool) + else None, + ), + trust=TrustClass.AGENT_GENERATED, + status=ExperienceStatus.SHADOW, + source=ExperienceSource( + provider="langgraph", + source_type="graph_state", + source_id=str(state.get("thread_id") or ""), + ), + ) + stored, decision = compiler.submit( + record, + context=FirewallContext(namespace=namespace, actor="langgraph"), + ) + return { + "wavemind_experience_id": stored.id, + "wavemind_firewall": decision.as_dict(), + } + + return capture_node diff --git a/wavemind/integrations/mcp_experience.py b/wavemind/integrations/mcp_experience.py new file mode 100644 index 00000000..a0e548eb --- /dev/null +++ b/wavemind/integrations/mcp_experience.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from wavemind.experience_compiler import ExperienceCompiler +from wavemind.memory_firewall import FirewallContext + + +class ExperienceMCPAdapter: + """Framework-neutral implementation of MCP tools/list and tools/call.""" + + def __init__(self, compiler: ExperienceCompiler): + self.compiler = compiler + + def list_tools(self) -> list[dict[str, Any]]: + return [ + { + "name": "compile_experience_packet", + "description": "Compile a trusted, token-bounded experience packet.", + "inputSchema": { + "type": "object", + "properties": { + "query": {"type": "string", "minLength": 1}, + "namespace": {"type": "string", "minLength": 1}, + "token_budget": {"type": "integer", "minimum": 32}, + "top_k": {"type": "integer", "minimum": 1, "maximum": 100}, + }, + "required": ["query", "namespace"], + "additionalProperties": False, + }, + }, + { + "name": "expand_experience", + "description": "Expand cited experience IDs with full provenance.", + "inputSchema": { + "type": "object", + "properties": { + "experience_ids": { + "type": "array", + "items": {"type": "string", "minLength": 1}, + "maxItems": 100, + }, + "namespace": {"type": "string", "minLength": 1}, + }, + "required": ["experience_ids", "namespace"], + "additionalProperties": False, + }, + }, + ] + + def call_tool( + self, + name: str, + arguments: Mapping[str, Any], + ) -> dict[str, Any]: + namespace = str(arguments.get("namespace") or "").strip() + if not namespace: + raise ValueError("namespace must not be empty") + context = FirewallContext(namespace=namespace, actor="mcp_client") + if name == "compile_experience_packet": + packet = self.compiler.compile_packet( + str(arguments.get("query") or ""), + namespace=namespace, + context=context, + token_budget=int(arguments.get("token_budget") or 800), + top_k=int(arguments.get("top_k") or 8), + ) + return packet.as_dict() + if name == "expand_experience": + ids = arguments.get("experience_ids") or [] + if not isinstance(ids, list): + raise ValueError("experience_ids must be an array") + return { + "schema": "wavemind.experience_details.v1", + "items": [ + detail.__dict__ + for detail in self.compiler.expand( + (str(item) for item in ids), + namespace=namespace, + context=context, + ) + ], + } + raise KeyError(name) + + +def build_experience_mcp_server( + compiler: ExperienceCompiler, + *, + name: str = "WaveMind Experience", +) -> Any: + try: + from mcp.server.fastmcp import FastMCP + except ImportError as exc: + raise RuntimeError( + 'MCP support requires: pip install "wavemind[mcp]"' + ) from exc + adapter = ExperienceMCPAdapter(compiler) + server = FastMCP(name=name, json_response=True) + + @server.tool(description="Compile a trusted, token-bounded experience packet.") + def compile_experience_packet( + query: str, + namespace: str, + token_budget: int = 800, + top_k: int = 8, + ) -> dict[str, Any]: + return adapter.call_tool( + "compile_experience_packet", + { + "query": query, + "namespace": namespace, + "token_budget": token_budget, + "top_k": top_k, + }, + ) + + @server.tool(description="Expand cited experience IDs with full provenance.") + def expand_experience( + experience_ids: list[str], + namespace: str, + ) -> dict[str, Any]: + return adapter.call_tool( + "expand_experience", + { + "experience_ids": experience_ids, + "namespace": namespace, + }, + ) + + return server diff --git a/wavemind/integrations/openai_agents.py b/wavemind/integrations/openai_agents.py new file mode 100644 index 00000000..9b8269af --- /dev/null +++ b/wavemind/integrations/openai_agents.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Awaitable, Callable, Mapping, Sequence +from pathlib import Path +from threading import RLock +from typing import Any + +from wavemind.experience_compiler import ExperienceCompiler +from wavemind.memory_firewall import FirewallContext + + +class WaveMindAgentsSession: + """Durable implementation of the OpenAI Agents SDK Session protocol.""" + + session_settings: Any | None = None + + def __init__( + self, + session_id: str, + *, + db_path: str | Path = "wavemind-agents.db", + ): + selected = str(session_id).strip() + if not selected: + raise ValueError("session_id must not be empty") + self.session_id = selected + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._lock = RLock() + self._conn = sqlite3.connect( + str(self.db_path), + check_same_thread=False, + timeout=30.0, + ) + self._conn.row_factory = sqlite3.Row + with self._conn: + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS openai_session_items ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + payload_json TEXT NOT NULL + ) + """ + ) + self._conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_openai_session_items + ON openai_session_items(session_id, sequence) + """ + ) + + async def get_items(self, limit: int | None = None) -> list[dict[str, Any]]: + if limit is not None and int(limit) < 1: + raise ValueError("limit must be positive") + query = ( + "SELECT payload_json FROM openai_session_items " + "WHERE session_id = ? ORDER BY sequence" + ) + values: list[Any] = [self.session_id] + if limit is not None: + query = ( + "SELECT payload_json FROM (" + "SELECT sequence, payload_json FROM openai_session_items " + "WHERE session_id = ? ORDER BY sequence DESC LIMIT ?" + ") ORDER BY sequence" + ) + values.append(int(limit)) + with self._lock: + rows = self._conn.execute(query, values).fetchall() + return [json.loads(str(row["payload_json"])) for row in rows] + + async def add_items(self, items: Sequence[Mapping[str, Any]]) -> None: + payloads = [_serialize_item(item) for item in items] + if not payloads: + return + with self._lock, self._conn: + self._conn.executemany( + """ + INSERT INTO openai_session_items (session_id, payload_json) + VALUES (?, ?) + """, + [(self.session_id, payload) for payload in payloads], + ) + + async def pop_item(self) -> dict[str, Any] | None: + with self._lock, self._conn: + row = self._conn.execute( + """ + SELECT sequence, payload_json FROM openai_session_items + WHERE session_id = ? ORDER BY sequence DESC LIMIT 1 + """, + (self.session_id,), + ).fetchone() + if row is None: + return None + self._conn.execute( + "DELETE FROM openai_session_items WHERE sequence = ?", + (int(row["sequence"]),), + ) + return json.loads(str(row["payload_json"])) + + async def clear_session(self) -> None: + with self._lock, self._conn: + self._conn.execute( + "DELETE FROM openai_session_items WHERE session_id = ?", + (self.session_id,), + ) + + def close(self) -> None: + with self._lock: + self._conn.close() + + def __enter__(self) -> "WaveMindAgentsSession": + return self + + def __exit__(self, *_: object) -> None: + self.close() + + +def make_experience_input_callback( + compiler: ExperienceCompiler, + *, + namespace: str, + token_budget: int = 800, + top_k: int = 8, + actor: str = "openai_agents", +) -> Callable[ + [Sequence[Mapping[str, Any]], Sequence[Mapping[str, Any]]], + Awaitable[list[dict[str, Any]]], +]: + """Build the official session_input_callback without persisting packet text.""" + + async def callback( + history: Sequence[Mapping[str, Any]], + new_input: Sequence[Mapping[str, Any]], + ) -> list[dict[str, Any]]: + query = _latest_text(new_input) + combined = [dict(item) for item in history] + if query: + packet = compiler.compile_packet( + query, + namespace=namespace, + context=FirewallContext(namespace=namespace, actor=actor), + token_budget=token_budget, + top_k=top_k, + ) + if packet.items: + combined.append( + { + "role": "system", + "content": packet.as_prompt(), + "metadata": { + "wavemind_schema": "wavemind.experience_packet.v1", + "citations": list(packet.citations), + "ephemeral": True, + }, + } + ) + combined.extend(dict(item) for item in new_input) + return combined + + return callback + + +def _serialize_item(item: Mapping[str, Any]) -> str: + try: + return json.dumps( + dict(item), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + except (TypeError, ValueError) as exc: + raise ValueError("session item must be JSON serializable") from exc + + +def _latest_text(items: Sequence[Mapping[str, Any]]) -> str: + for item in reversed(items): + content = item.get("content") + if isinstance(content, str) and content.strip(): + return content.strip() + if isinstance(content, Sequence) and not isinstance(content, (str, bytes)): + parts = [ + str(part.get("text")) + for part in content + if isinstance(part, Mapping) and part.get("text") + ] + if parts: + return "\n".join(parts) + return ""