diff --git a/docs/AGENT-SETUP.md b/docs/AGENT-SETUP.md index c57f0a8c..c39b499e 100644 --- a/docs/AGENT-SETUP.md +++ b/docs/AGENT-SETUP.md @@ -63,6 +63,8 @@ The package has two paths: - **HTTP event capture**: the Pi extension sends prompts, summaries, passive task learnings, and compact Pi-native `mem_*` tool calls to `engram serve`. - **MCP gateway**: `pi-mcp-adapter` exposes Engram's MCP surface by launching `engram mcp --tools=agent` and is also used by other Pi MCP integrations such as Notion. +Pi-native `mem_save`, `mem_save_prompt`, `mem_session_summary`, and `mem_capture_passive` calls use only Pi's current `ctx.sessionManager.getSessionId()` for session attribution. Their tool schemas do not accept `session_id`; if the runtime ID is missing or session registration is not acknowledged, the extension stops before sending the attributed write. + Use an existing Engram HTTP server: ```bash @@ -86,8 +88,6 @@ If the binary is missing, the MCP launcher exits cleanly instead of crashing Pi Other write tools still primarily use cwd/repo detection unless their schema says otherwise. Start the MCP server from the repo or add `.engram/config.json` when you want deterministic default writes. -OpenCode binds `mem_save`, `mem_save_prompt`, `mem_session_summary`, and `mem_capture_passive` to its confirmed top-level runtime session and maps subagents to their authoritative parent. - To lock write tools to the canonical project for a repo, add `.engram/config.json` at the repo root: ```json @@ -221,6 +221,8 @@ This does three things: The plugin auto-starts the HTTP server if needed for session tracking. If your environment blocks background processes, run it manually: +OpenCode binds `mem_save`, `mem_save_prompt`, `mem_session_summary`, and `mem_capture_passive` to its confirmed top-level runtime session and maps subagents to their authoritative parent. + ```bash engram serve & ``` diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index f212f39b..36188f09 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -47,6 +47,8 @@ Session ends → Agent writes session summary (Goal/Discoveries/Accomplished/Nex Next session starts → Previous session context is injected automatically ``` +Host adapters translate authoritative runtime identity at their boundary; durable session lifecycle and persistence semantics remain in the Go core. OpenCode maps attributed writes to its confirmed top-level runtime session, including authoritative parent mapping for subagents. Pi binds its four native session-attributed writes (`mem_save`, `mem_save_prompt`, `mem_session_summary`, and `mem_capture_passive`) to the current Pi `SessionContext` ID and stops before the write when that ID or its session-registration acknowledgement is unavailable. + --- ## MCP Tools diff --git a/docs/PLUGINS.md b/docs/PLUGINS.md index 7e799358..b7400681 100644 --- a/docs/PLUGINS.md +++ b/docs/PLUGINS.md @@ -7,6 +7,7 @@ > Validation boundary (current): plugin scripts are validated for memory/session workflows, not as cloud bootstrap orchestrators. Use CLI for cloud config/auth/enrollment/upgrade. - [Current plugin coverage](#current-plugin-coverage) +- [Pi Extension](#pi-extension) - [OpenCode Plugin](#opencode-plugin) - [Claude Code Plugin](#claude-code-plugin) - [Privacy](#privacy) @@ -24,6 +25,18 @@ --- +## Pi Extension + +For Pi users, the `gentle-engram` package is a thin adapter over `engram serve`. It captures Pi lifecycle events, injects the Memory Protocol, and exposes compact Pi-native `mem_*` tools over HTTP. The optional `pi-mcp-adapter` path launches `engram mcp --tools=agent` separately for MCP integrations. + +The four Pi-native session-attributed writes—`mem_save`, `mem_save_prompt`, `mem_session_summary`, and `mem_capture_passive`—derive their session ID exclusively from Pi's current `SessionContext`. These schemas do not expose a model-supplied `session_id`. Before forwarding one of these writes, the extension registers that runtime session with Engram and requires an acknowledgement; a missing runtime ID or failed registration stops the write and leaves registration retryable. + +Project selection remains separate from host session identity. The adapter asks `engram serve` for canonical project detection while the Go core continues to own durable session, project, and persistence semantics. + +See [`plugin/pi/README.md`](../plugin/pi/README.md) for installation, configuration, and troubleshooting. + +--- + ## OpenCode Plugin For [OpenCode](https://opencode.ai) users, a thin TypeScript plugin adds enhanced session management on top of the MCP tools: diff --git a/plugin/pi/README.md b/plugin/pi/README.md index abf19085..4a3c33f1 100644 --- a/plugin/pi/README.md +++ b/plugin/pi/README.md @@ -101,6 +101,12 @@ Pi MCP tools -> pi-mcp-adapter -> ENGRAM_BIN / engram mcp -> SQLite Pi-native compact tools use the same HTTP server path as event capture, including project detection, diagnostics, passive capture, lifecycle review, and conflict-judgment tools such as `mem_current_project`, `mem_doctor`, `mem_capture_passive`, `mem_review`, `mem_judge`, and `mem_compare`. MCP tools remain a separate stdio path, so direct MCP usage still needs an Engram binary even when `ENGRAM_URL` points at a remote HTTP server. Engram MCP direct tools are not enabled by default in Pi to avoid duplicate raw `engram_mem_*` tool rows. +## Runtime session attribution + +Pi-native `mem_save`, `mem_save_prompt`, `mem_session_summary`, and `mem_capture_passive` calls are attributed to the current Pi runtime session. The extension reads the ID from Pi's `SessionContext`; these tool schemas do not accept a model-supplied `session_id`. + +Before forwarding any of these four writes, the extension registers the runtime session with `engram serve` and requires an acknowledgement. If Pi has no current runtime ID, or if registration cannot be confirmed, the attributed write is not sent. Failed registration is not cached, so a later call can retry safely. Engram's Go core remains responsible for durable session, project, and persistence semantics. + ## Compact memory tool rendering `gentle-engram` owns the Pi chrome for Engram memory tools by registering compact Pi-native `mem_*` tools in the companion package. When tools such as `mem_search`, `mem_context`, `mem_save`, `mem_session_summary`, `mem_get_observation`, `mem_review`, `mem_judge`, and `mem_doctor` run in Pi, the default collapsed view stays compact: diff --git a/plugin/pi/index.ts b/plugin/pi/index.ts index 33a5523d..50fae670 100644 --- a/plugin/pi/index.ts +++ b/plugin/pi/index.ts @@ -178,10 +178,9 @@ function isTimeoutError(error: unknown): boolean { return error instanceof Error && (error.name === "TimeoutError" || error.name === "AbortError"); } -// engramFetch resolves to null on failure and ~20 call sites depend on that fallthrough — -// ensureSession in particular must not abort a mem_save just because session creation blipped. -// So the timeout detail travels out-of-band instead of changing what any caller receives, -// letting executeMemoryTool tell the truth about an ambiguous write without blast radius. +// engramFetch resolves to null on transport failure. Session-attributed writes +// treat a null registration response as unacknowledged and stop before writing; +// other callers retain the existing null fallthrough contract. let lastFetchTimeoutMethod: string | undefined; function takeLastFetchTimeoutMethod(): string | undefined { @@ -415,14 +414,33 @@ let projectResolutionError: string | undefined; let projectDetectionPending = false; const knownSessions = new Set(); +const sessionRegistrationsInFlight = new Map>(); const toolCounts = new Map(); async function ensureSession(sessionId: string, sessionProject = project): Promise { const key = `${sessionProject}:${sessionId}`; if (!sessionId || knownSessions.has(key)) return; - knownSessions.add(key); - const body: SessionBody = { id: sessionId, project: sessionProject, directory }; - await engramFetch("/sessions", { method: "POST", body }); + + const existingRegistration = sessionRegistrationsInFlight.get(key); + if (existingRegistration) return existingRegistration; + + const registration = (async () => { + const body: SessionBody = { id: sessionId, project: sessionProject, directory }; + const acknowledgement = await engramFetch("/sessions", { method: "POST", body }); + if (acknowledgement === null) { + throw new Error(`gentle-engram could not confirm session registration for Pi runtime session ${sessionId}`); + } + knownSessions.add(key); + })(); + sessionRegistrationsInFlight.set(key, registration); + + try { + await registration; + } finally { + if (sessionRegistrationsInFlight.get(key) === registration) { + sessionRegistrationsInFlight.delete(key); + } + } } async function detectServerProject(cwd: string): Promise { @@ -507,6 +525,14 @@ function getSessionId(ctx: SessionContext): string | undefined { return ctx.sessionManager.getSessionId(); } +function requireRuntimeSessionID(ctx: SessionContext): string { + const sessionId = ctx.sessionManager.getSessionId()?.trim(); + if (!sessionId) { + throw new Error("Pi runtime session ID is unavailable; session-attributed writes require a native SessionContext ID"); + } + return sessionId; +} + const optionalString = (description: string) => Type.Optional(Type.String({ description })); const optionalNumber = (description: string) => Type.Optional(Type.Number({ description })); const optionalBoolean = (description: string) => Type.Optional(Type.Boolean({ description })); @@ -525,7 +551,6 @@ const MEMORY_TOOL_SCHEMAS: Record> = { title: Type.String({ description: "Short, searchable title" }), content: Type.String({ description: "Structured memory content" }), type: optionalString("Observation type/category"), - session_id: optionalString("Session ID to associate with"), scope: optionalString("Scope: project or personal"), topic_key: optionalString("Stable topic key for upserts"), project: optionalString("Optional explicit project"), @@ -550,12 +575,10 @@ const MEMORY_TOOL_SCHEMAS: Record> = { }), mem_save_prompt: Type.Object({ content: Type.String({ description: "The user's prompt text" }), - session_id: optionalString("Session ID to associate with"), project: optionalString("Optional project"), }), mem_session_summary: Type.Object({ content: Type.String({ description: "Full session summary" }), - session_id: optionalString("Session ID"), project: optionalString("Optional project to use when automatic detection is unavailable"), }), mem_context: Type.Object({ @@ -591,7 +614,6 @@ const MEMORY_TOOL_SCHEMAS: Record> = { }), mem_capture_passive: Type.Object({ content: Type.String({ description: "Text output containing a ## Key Learnings section" }), - session_id: optionalString("Session ID to associate with"), source: optionalString("Source identifier, e.g. subagent-stop or session-end"), }), mem_review: Type.Object({ @@ -652,7 +674,7 @@ async function callMemoryTool(toolName: string, params: Record, const sessionId = getSessionId(ctx); const requestedProject = typeof params.project === "string" && params.project ? params.project : undefined; const activeProject = requestedProject || project; - const activeSessionId = String(params.session_id || (requestedProject ? `manual-save-${requestedProject}` : sessionId) || `manual-save-${project}`); + const runtimeSessionForWrite = () => requireRuntimeSessionID(ctx); switch (toolName) { case "mem_search": @@ -674,8 +696,9 @@ async function callMemoryTool(toolName: string, params: Record, return engramFetch(`/timeline${queryString({ observation_id: params.observation_id, before: params.before, after: params.after, project: params.project })}`); case "mem_get_observation": return engramFetch(`/observations/${encodeURIComponent(String(params.id))}`); - case "mem_save": + case "mem_save": { if (!requestedProject) requireResolvedProject(); + const activeSessionId = runtimeSessionForWrite(); await ensureSession(activeSessionId, activeProject); return engramFetch("/observations", { method: "POST", @@ -689,6 +712,7 @@ async function callMemoryTool(toolName: string, params: Record, topic_key: params.topic_key, }, }); + } case "mem_update": return engramFetch(`/observations/${encodeURIComponent(String(params.id))}`, { method: "PATCH", @@ -704,20 +728,23 @@ async function callMemoryTool(toolName: string, params: Record, return engramFetch(`/observations/${encodeURIComponent(String(params.id))}${queryString({ hard: params.hard_delete })}`, { method: "DELETE" }); case "mem_suggest_topic_key": return { topic_key: slugifyTopicKey(params) }; - case "mem_save_prompt": + case "mem_save_prompt": { if (!requestedProject) requireResolvedProject(); - await ensureSession(activeSessionId, activeProject); + const promptSessionId = runtimeSessionForWrite(); + await ensureSession(promptSessionId, activeProject); return engramFetch("/prompts", { method: "POST", - body: { session_id: activeSessionId, content: params.content, project: activeProject }, + body: { session_id: promptSessionId, content: params.content, project: activeProject }, }); - case "mem_session_summary": + } + case "mem_session_summary": { if (!requestedProject) requireResolvedProject(); - await ensureSession(activeSessionId, activeProject); + const summarySessionId = runtimeSessionForWrite(); + await ensureSession(summarySessionId, activeProject); return engramFetch("/observations", { method: "POST", body: { - session_id: activeSessionId, + session_id: summarySessionId, type: "session_summary", title: "Session summary", content: params.content, @@ -725,6 +752,7 @@ async function callMemoryTool(toolName: string, params: Record, scope: "project", }, }); + } case "mem_session_start": requireResolvedProject(); return engramFetch("/sessions", { @@ -749,18 +777,20 @@ async function callMemoryTool(toolName: string, params: Record, } case "mem_doctor": return engramFetch(`/doctor${queryString({ project: params.project, check: params.check, cwd: params.project ? undefined : ctx.cwd })}`); - case "mem_capture_passive": + case "mem_capture_passive": { requireResolvedProject(); - await ensureSession(activeSessionId); + const passiveSessionId = runtimeSessionForWrite(); + await ensureSession(passiveSessionId); return engramFetch("/observations/passive", { method: "POST", body: { - session_id: activeSessionId, + session_id: passiveSessionId, content: params.content, project, source: params.source || "pi-tool", }, }); + } case "mem_review": { const action = String(params.action || "").trim(); if (action === "list") { diff --git a/plugin/pi/test/index-source.test.mjs b/plugin/pi/test/index-source.test.mjs index 4246eaae..22cf8504 100644 --- a/plugin/pi/test/index-source.test.mjs +++ b/plugin/pi/test/index-source.test.mjs @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { readFileSync } from "node:fs"; import { test } from "node:test"; -const source = readFileSync(new URL("../index.ts", import.meta.url), "utf8"); +const source = readFileSync(new URL("../index.ts", import.meta.url), "utf8").replaceAll("\r\n", "\n"); function extractFunctionBody(name, marker) { const signatureIndex = source.indexOf(`function ${name}`); @@ -103,6 +103,22 @@ function buildScheduleEngramSelfHealForTest({ waitUnref, isEngramRunning, maxAtt return factory(waitUnref, isEngramRunning, 1, maxAttempts); } +function buildEnsureSessionForTest(engramFetch) { + const body = extractFunctionBody("ensureSession", "{\n const key") + .replace("const body: SessionBody", "const body"); + const factory = new Function("knownSessions", "sessionRegistrationsInFlight", "engramFetch", "project", "directory", ` + return async function ensureSession(sessionId, sessionProject = project) { + ${body} + }; + `); + const knownSessions = new Set(); + const sessionRegistrationsInFlight = new Map(); + return { + ensureSession: factory(knownSessions, sessionRegistrationsInFlight, engramFetch, "engram", "/work/engram"), + knownSessions, + }; +} + function sessionCtx(id, sink) { return { sessionManager: { getSessionId: () => id }, @@ -112,7 +128,7 @@ function sessionCtx(id, sink) { test("mem_session_summary accepts explicit project fallback", () => { assert.match(source, /mem_session_summary: Type\.Object\(\{[\s\S]*project: optionalString\("Optional project to use when automatic detection is unavailable"\)/); - assert.match(source, /case "mem_session_summary":[\s\S]*if \(!requestedProject\) requireResolvedProject\(\);[\s\S]*ensureSession\(activeSessionId, activeProject\)[\s\S]*project: activeProject/); + assert.match(source, /case "mem_session_summary":[\s\S]*if \(!requestedProject\) requireResolvedProject\(\);[\s\S]*ensureSession\(summarySessionId, activeProject\)[\s\S]*project: activeProject/); }); test("mem_search exposes and forwards match_mode and all_projects", () => { @@ -286,35 +302,32 @@ test("the tool layer reports unknown write outcome instead of inviting a blind r assert.doesNotMatch(unreachable, /timed out/); }); -test("a session-creation timeout still lets the observation write through", async () => { - // Regression: when engramFetch threw on timeout, the unguarded ensureSession call in - // mem_save aborted the whole tool call before /observations was ever attempted, silently - // dropping the user's memory while telling the agent not to retry. - assert.match(source, /await ensureSession\(activeSessionId, activeProject\);/); - assert.doesNotMatch(source, /throw new EngramTimeoutError/); +test("session registration requires acknowledgement and failed acknowledgement remains retryable", async () => { + let calls = 0; + const { ensureSession, knownSessions } = buildEnsureSessionForTest(async () => { + calls += 1; + return calls === 1 ? null : { status: "created" }; + }); - const originalFetch = globalThis.fetch; - const paths = []; - globalThis.fetch = async (url, init) => { - const path = new URL(url).pathname; - paths.push(path); - if (path === "/sessions") { - const timeout = new Error("The operation was aborted due to timeout"); - timeout.name = "TimeoutError"; - throw timeout; - } - return { ok: true, async json() { return { id: 1 }; } }; - }; - try { - const { engramFetch } = buildEngramFetchForTest(); - // ensureSession's own call fails soft... - assert.equal(await engramFetch("/sessions", { method: "POST", body: { id: "s" } }), null); - // ...and the observation write that follows it still lands. - assert.deepEqual(await engramFetch("/observations", { method: "POST", body: { title: "t" } }), { id: 1 }); - assert.deepEqual(paths, ["/sessions", "/observations"]); - } finally { - globalThis.fetch = originalFetch; + await assert.rejects(ensureSession("runtime"), /could not confirm session registration/); + assert.equal(knownSessions.has("engram:runtime"), false); + await ensureSession("runtime"); + assert.equal(knownSessions.has("engram:runtime"), true); + await ensureSession("runtime"); + assert.equal(calls, 2); +}); + +test("four session-attributed writes ignore model session_id and require the Pi runtime ID", () => { + for (const tool of ["mem_save", "mem_save_prompt", "mem_session_summary", "mem_capture_passive"]) { + const schema = source.match(new RegExp(`${tool}: Type\\.Object\\(\\{([\\s\\S]*?)\\n \\}\\),`)); + assert.ok(schema, `${tool} schema not found`); + assert.doesNotMatch(schema[1], /session_id:/, `${tool} must not invite model-supplied session identity`); } + assert.match(source, /function requireRuntimeSessionID/); + assert.match(source, /ctx\.sessionManager\.getSessionId\(\)/); + assert.match(source, /Pi runtime session ID is unavailable/); + assert.doesNotMatch(source, /const activeSessionId = String\(params\.session_id/); + assert.doesNotMatch(source, /manual-save-\$\{requestedProject\}/); }); test("a timeout on the session leg does not mislabel an unrelated failure on the write leg", async () => { diff --git a/plugin/pi/test/native-tool-contract.test.mjs b/plugin/pi/test/native-tool-contract.test.mjs index 459273e2..e7aeb3bc 100644 --- a/plugin/pi/test/native-tool-contract.test.mjs +++ b/plugin/pi/test/native-tool-contract.test.mjs @@ -31,6 +31,40 @@ export const Type = new Proxy({}, { get: (_target, prop) => schema(String(prop)) ); } +function deferred() { + let resolve; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + +async function loadPluginHarness(tag) { + await installRuntimeStubs(); + const registeredTools = new Map(); + const eventHandlers = new Map(); + const pluginUrl = pathToFileURL(join(ROOT, "index.ts")); + pluginUrl.search = `?${tag}=${Date.now()}`; + const { default: registerEngram } = await import(pluginUrl.href); + registerEngram({ + registerTool(tool) { + registeredTools.set(tool.name, tool); + }, + on(event, handler) { + eventHandlers.set(event, handler); + }, + }); + return { registeredTools, eventHandlers }; +} + +function runtimeContext(sessionId) { + return { + cwd: ROOT, + sessionManager: { getSessionId: () => sessionId }, + ui: { setStatus() {} }, + }; +} + test("registered Pi-native mem_search reports native provider transport failure", async () => { const originalFetch = globalThis.fetch; const originalUrl = process.env.ENGRAM_URL; @@ -78,3 +112,250 @@ test("registered Pi-native mem_search reports native provider transport failure" await rm(NODE_MODULES, { recursive: true, force: true }); } }); + +test("all session-attributed Pi writes bind to acknowledged runtime identity and fail closed", async () => { + const originalFetch = globalThis.fetch; + const originalUrl = process.env.ENGRAM_URL; + process.env.ENGRAM_URL = "http://127.0.0.1:17437"; + const registrationAttempts = new Map(); + const sessionBodies = []; + const writeRequests = []; + globalThis.fetch = async (url, init) => { + const path = new URL(url).pathname; + if (path === "/health") return { ok: true, async json() { return { status: "ok" }; } }; + if (path === "/project/current") { + return { ok: true, async json() { return { project: "pi", project_source: "dir_basename", project_path: ROOT }; } }; + } + if (path === "/sessions") { + const body = JSON.parse(init.body); + const attempts = (registrationAttempts.get(body.id) || 0) + 1; + registrationAttempts.set(body.id, attempts); + sessionBodies.push(body); + if (attempts === 1) { + return { ok: false, status: 503, async json() { return { error: "registration unavailable" }; } }; + } + return { ok: true, status: 201, async json() { return { status: "created" }; } }; + } + if (["/observations", "/prompts", "/observations/passive"].includes(path)) { + writeRequests.push({ path, body: JSON.parse(init.body) }); + return { ok: true, status: 201, async json() { return { id: writeRequests.length }; } }; + } + return { ok: true, async json() { return {}; } }; + }; + + try { + await installRuntimeStubs(); + const registeredTools = new Map(); + const pluginUrl = pathToFileURL(join(ROOT, "index.ts")); + pluginUrl.search = `?binding=${Date.now()}`; + const { default: registerEngram } = await import(pluginUrl.href); + registerEngram({ + registerTool(tool) { registeredTools.set(tool.name, tool); }, + on() {}, + }); + + const cases = [ + { + name: "mem_save", + path: "/observations", + params: { title: "runtime binding", content: "content", session_id: "model-invented" }, + assertBody(body) { + assert.equal(body.title, "runtime binding"); + assert.equal(body.type, "manual"); + }, + }, + { + name: "mem_save_prompt", + path: "/prompts", + params: { content: "prompt", session_id: "model-invented" }, + assertBody(body) { + assert.equal(body.content, "prompt"); + }, + }, + { + name: "mem_session_summary", + path: "/observations", + params: { content: "summary", session_id: "model-invented" }, + assertBody(body) { + assert.equal(body.type, "session_summary"); + assert.equal(body.content, "summary"); + }, + }, + { + name: "mem_capture_passive", + path: "/observations/passive", + params: { content: "## Key Learnings:\n- Runtime identity", source: "contract-test", session_id: "model-invented" }, + assertBody(body) { + assert.equal(body.source, "contract-test"); + }, + }, + ]; + + for (const contract of cases) { + const tool = registeredTools.get(contract.name); + assert.ok(tool, `${contract.name} tool should be registered`); + const runtimeSession = `runtime-${contract.name}`; + const ctx = { + cwd: ROOT, + sessionManager: { getSessionId: () => runtimeSession }, + ui: { setStatus() {} }, + }; + const writesBefore = writeRequests.length; + + const failed = await tool.execute(`${contract.name}-failed`, contract.params, undefined, undefined, ctx); + assert.equal(failed.isError, true, `${contract.name} should report failed registration`); + assert.equal(writeRequests.length, writesBefore, `${contract.name} must stop before an unacknowledged write`); + + const succeeded = await tool.execute(`${contract.name}-success`, contract.params, undefined, undefined, ctx); + assert.equal(succeeded.isError, undefined, `${contract.name} should succeed after registration is acknowledged`); + assert.equal(registrationAttempts.get(runtimeSession), 2, `${contract.name} registration failure must remain retryable`); + const forwarded = writeRequests.at(-1); + assert.equal(forwarded.path, contract.path); + assert.equal(forwarded.body.session_id, runtimeSession); + assert.notEqual(forwarded.body.session_id, contract.params.session_id); + assert.equal(forwarded.body.project, "pi"); + contract.assertBody(forwarded.body); + + await tool.execute(`${contract.name}-cached`, contract.params, undefined, undefined, ctx); + assert.equal(registrationAttempts.get(runtimeSession), 2, `${contract.name} should cache acknowledged registration`); + + const registrationsBeforeMissingRuntime = sessionBodies.length; + const writesBeforeMissingRuntime = writeRequests.length; + const noRuntime = await tool.execute( + `${contract.name}-missing-runtime`, + contract.params, + undefined, + undefined, + { ...ctx, sessionManager: { getSessionId: () => undefined } }, + ); + assert.equal(noRuntime.isError, true); + assert.match(noRuntime.content[0].text, /Pi runtime session ID is unavailable/); + assert.equal(sessionBodies.length, registrationsBeforeMissingRuntime, `${contract.name} must not register a synthetic session`); + assert.equal(writeRequests.length, writesBeforeMissingRuntime, `${contract.name} must not write without runtime identity`); + } + } finally { + globalThis.fetch = originalFetch; + if (originalUrl === undefined) delete process.env.ENGRAM_URL; + else process.env.ENGRAM_URL = originalUrl; + await rm(NODE_MODULES, { recursive: true, force: true }); + } +}); + +test("parallel first-use writes share one acknowledged registration and keep it cached", async () => { + const originalFetch = globalThis.fetch; + const originalUrl = process.env.ENGRAM_URL; + process.env.ENGRAM_URL = "http://127.0.0.1:17437"; + const registrationGate = deferred(); + let registrationAttempts = 0; + const writeRequests = []; + globalThis.fetch = async (url, init) => { + const path = new URL(url).pathname; + if (path === "/health") return { ok: true, async json() { return { status: "ok" }; } }; + if (path === "/project/current") { + return { ok: true, async json() { return { project: "pi", project_source: "dir_basename", project_path: ROOT }; } }; + } + if (path === "/sessions") { + registrationAttempts += 1; + await registrationGate.promise; + return { ok: true, status: 201, async json() { return { status: "created" }; } }; + } + if (path === "/observations") { + writeRequests.push(JSON.parse(init.body)); + return { ok: true, status: 201, async json() { return { id: writeRequests.length }; } }; + } + return { ok: true, async json() { return {}; } }; + }; + + try { + const { registeredTools, eventHandlers } = await loadPluginHarness("parallel-success"); + const memSave = registeredTools.get("mem_save"); + const ctx = runtimeContext("parallel-success-session"); + await eventHandlers.get("session_start")({}, ctx); + + const firstWrite = memSave.execute("parallel-success-1", { title: "first", content: "one" }, undefined, undefined, ctx); + const secondWrite = memSave.execute("parallel-success-2", { title: "second", content: "two" }, undefined, undefined, ctx); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(registrationAttempts, 1, "parallel first writes must share one registration request"); + + registrationGate.resolve(); + const [firstResult, secondResult] = await Promise.all([firstWrite, secondWrite]); + assert.equal(firstResult.isError, undefined); + assert.equal(secondResult.isError, undefined); + assert.deepEqual(writeRequests.map((request) => request.title).sort(), ["first", "second"]); + assert.ok(writeRequests.every((request) => request.session_id === "parallel-success-session")); + + await memSave.execute("parallel-success-cached", { title: "cached", content: "three" }, undefined, undefined, ctx); + assert.equal(registrationAttempts, 1, "acknowledged registration must remain cached"); + assert.equal(writeRequests.length, 3); + } finally { + globalThis.fetch = originalFetch; + if (originalUrl === undefined) delete process.env.ENGRAM_URL; + else process.env.ENGRAM_URL = originalUrl; + await rm(NODE_MODULES, { recursive: true, force: true }); + } +}); + +test("shared registration failure rejects parallel writes and a later call retries", async () => { + const originalFetch = globalThis.fetch; + const originalUrl = process.env.ENGRAM_URL; + process.env.ENGRAM_URL = "http://127.0.0.1:17437"; + const registrationGate = deferred(); + let registrationAttempts = 0; + let registrationShouldFail = true; + const writeRequests = []; + globalThis.fetch = async (url, init) => { + const path = new URL(url).pathname; + if (path === "/health") return { ok: true, async json() { return { status: "ok" }; } }; + if (path === "/project/current") { + return { ok: true, async json() { return { project: "pi", project_source: "dir_basename", project_path: ROOT }; } }; + } + if (path === "/sessions") { + registrationAttempts += 1; + if (registrationShouldFail) { + await registrationGate.promise; + return { ok: false, status: 503, async json() { return { error: "registration unavailable" }; } }; + } + return { ok: true, status: 201, async json() { return { status: "created" }; } }; + } + if (path === "/observations") { + writeRequests.push(JSON.parse(init.body)); + return { ok: true, status: 201, async json() { return { id: writeRequests.length }; } }; + } + return { ok: true, async json() { return {}; } }; + }; + + try { + const { registeredTools, eventHandlers } = await loadPluginHarness("parallel-failure"); + const memSave = registeredTools.get("mem_save"); + const ctx = runtimeContext("parallel-failure-session"); + await eventHandlers.get("session_start")({}, ctx); + + const firstWrite = memSave.execute("parallel-failure-1", { title: "first", content: "one" }, undefined, undefined, ctx); + const secondWrite = memSave.execute("parallel-failure-2", { title: "second", content: "two" }, undefined, undefined, ctx); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(registrationAttempts, 1, "parallel failed writes must share one registration request"); + + registrationGate.resolve(); + const [firstResult, secondResult] = await Promise.all([firstWrite, secondWrite]); + assert.equal(firstResult.isError, true); + assert.equal(secondResult.isError, true); + assert.match(firstResult.content[0].text, /registration unavailable/); + assert.match(secondResult.content[0].text, /registration unavailable/); + assert.equal(writeRequests.length, 0, "failed registration must stop every waiting write"); + + registrationShouldFail = false; + const retryResult = await memSave.execute("parallel-failure-retry", { title: "retry", content: "three" }, undefined, undefined, ctx); + assert.equal(retryResult.isError, undefined); + assert.equal(registrationAttempts, 2, "a later write must retry failed registration"); + assert.equal(writeRequests.length, 1); + + await memSave.execute("parallel-failure-cached", { title: "cached", content: "four" }, undefined, undefined, ctx); + assert.equal(registrationAttempts, 2, "successful retry must remain cached"); + assert.equal(writeRequests.length, 2); + } finally { + globalThis.fetch = originalFetch; + if (originalUrl === undefined) delete process.env.ENGRAM_URL; + else process.env.ENGRAM_URL = originalUrl; + await rm(NODE_MODULES, { recursive: true, force: true }); + } +});