fix(pi): bind writes to acknowledged runtime sessions

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 &
 ```

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

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:

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:

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<string>();
+const sessionRegistrationsInFlight = new Map<string, Promise<void>>();
 const toolCounts = new Map<string, number>();
 
 async function ensureSession(sessionId: string, sessionProject = project): Promise<void> {
   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<CurrentProjectResponse | undefined> {
@@ -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<string, ReturnType<typeof Type.Object>> = {
     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<string, ReturnType<typeof Type.Object>> = {
   }),
   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<string, ReturnType<typeof Type.Object>> = {
   }),
   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<string, unknown>,
   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<string, unknown>,
       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<string, unknown>,
           topic_key: params.topic_key,
         },
       });
+    }
     case "mem_update":
       return engramFetch(`/observations/${encodeURIComponent(String(params.id))}`, {
         method: "PATCH",
@@ -704,27 +728,31 @@ async function callMemoryTool(toolName: string, params: Record<string, unknown>,
       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,
           project: activeProject,
           scope: "project",
         },
       });
+    }
     case "mem_session_start":
       requireResolvedProject();
       return engramFetch("/sessions", {
@@ -749,18 +777,20 @@ async function callMemoryTool(toolName: string, params: Record<string, unknown>,
     }
     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") {

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 () => {