fix(pi-plugin): support pi-web multi-session and RPC hosts

packages/pi-plugin/PARITY.md

@@ -23,14 +23,20 @@ plugin process and reach `experimental.chat.messages.transform`. OpenCode gates
 historian / m[0]m[1] injection / nudges / auto-search behind `fullFeatureMode`
 (i.e. `!isSubagent`), and detects subagents via OpenCode's `session.parent_id`.
 
-**Pi:** Pi has **no native subagent concept**. The *only* subagents that exist
-are the ones Magic Context itself spawns (historian, dreamer, sidekick), and each
-runs as a **separate `pi --print` process** loading only the lean
-`subagent-entry.js`, whose recursion guard **never wires `pi.on("context")`**
-(see `subagent-entry.ts` header). A Pi subagent therefore *cannot* reach the
-context-handler pipeline at all.
-
-**Consequence:** `is_subagent` is **never written `true`** for any Pi session.
+**Pi:** Pi has **no native subagent concept**. The subagents Magic Context itself
+spawns (historian, dreamer, sidekick) each run as a **separate `pi --print` process**
+loading only the lean `subagent-entry.js`, whose recursion guard **never wires
+`pi.on("context")`** (see `subagent-entry.ts` header). A Magic Context subagent
+therefore *cannot* reach the context-handler pipeline at all.
+`@gotgenes/pi-subagents`, however, can initialize a child session inside the same
+process. The full extension uses Pi's public child-session lifecycle events plus
+process-shared `AsyncLocalStorage` to suppress only the child while allowing
+unrelated same-process sessions to initialize normally.
+
+**Consequence:** `is_subagent` is **never written `true`** for any Pi session
+that reaches the context-handler pipeline. Separate child processes load the lean
+entry, while in-process child initialization is suppressed before the normal
+context pipeline is registered.
 There is nothing to gate, so Pi does NOT need OpenCode's `fullFeatureMode`
 reduced-mode enforcement in `context-handler.ts`. The vestigial `!isSubagent`
 checks that exist in the Pi context handler are harmless (always take the
@@ -115,15 +121,20 @@ the source array for dirty indices only.
 
 ---
 
-## 6. Transient UI: Pi uses `ctx.ui.notify` toasts, not persistent dialogs
+## 6. Transient UI: Pi uses `ctx.ui.notify` toasts and RPC dialogs
 
 **OpenCode:** TUI dialogs (upgrade prompt, `/ctx-status`, `/ctx-recomp`, `/ctx-embed`, `/ctx-flush`) via RPC,
 with an ignored-message fallback for Desktop/Web. Notification drain is
 **session-scoped** (a notification tagged for one session never surfaces in
 another) because one process can serve multiple sessions and TUI port discovery
 is newest-pid-wins.
 
-**Pi:** transient terminal notifications. The upgrade reminder passes
+**Pi:** command status is appended as a model-invisible custom entry. Interactive
+terminals render that entry through the registered entry renderer. In Pi RPC
+mode, each command uses its live `ctx`: `ctx.ui.notify` presents short progress
+as toasts and `ctx.ui.custom` presents detailed results as dialogs. A context
+captured by `session_start` cannot be reused because pi-web can host multiple
+sessions in one process. The upgrade reminder passes
 `deliveryPersists=false` on Pi, so a missed toast does not honor the old explicit-
 dismissal stamp. Both harnesses persist the 24-hour reminder cooldown and three-
 delivery cap, preventing repeated startup toasts while `/ctx-status` still reports
@@ -155,6 +166,13 @@ shared resolver's log-only dubious-ownership warning while still using the same
   **stdin** (Pi concatenates stdin + positional) to avoid Linux `MAX_ARG_STRLEN`
   / E2BIG; the positional is omitted when piping.
 - `--no-session` keeps subagent JSONL out of the user's session picker.
+- In pi-web, multiple sessions can share one process. Startup maintenance runs
+  once per process, while each session wires its own hooks. Dreamer registration
+  is process-shared and tracks sibling ownership, so one session's shutdown cannot
+  deregister another session's project timer.
+- `session_shutdown` drains only that session's in-flight historian and recomp work
+  and only the shutting-down extension instance's Dreamer work. Child-session
+  lifecycle listeners are detached only for that extension instance.
 
 ---
 
@@ -378,7 +396,8 @@ mechanism differs because the process models differ:
   inline `await` froze all input. Pi instead spawns the recomp via
   `spawnPiRecompRun` (mirroring `spawnPiHistorianRun`): the handler returns
   immediately after the ack message, the run is tracked in an in-flight map for
-  `session_shutdown` drain, and progress surfaces through `[ctx-status]`
+  `session_shutdown` drain (keyed by session id so one session does not drain
+  another), and progress surfaces through `[ctx-status]`
   messages + the `recomp` status-line flag.
 
 Because Pi's recomp runs in the background (not inside the user's turn), its

packages/pi-plugin/src/agent-end-handler.test.ts

@@ -107,15 +107,36 @@ describe("session_shutdown handler (drain location)", () => {
 	const body = extractSessionShutdownHandlerBody(INDEX_SRC);
 
 	test("drains in-flight historians through withTimeout", () => {
-		expect(body).toContain("awaitInFlightHistorians");
-		expect(body).toContain(
-			"withTimeout(awaitInFlightHistorians(), SHUTDOWN_DRAIN_MS)",
+		expect(body).toMatch(
+			/withTimeout\(\s*awaitInFlightHistorians\(sessionId\),\s*SHUTDOWN_DRAIN_MS,?\s*\)/,
 		);
 		expect(body).not.toContain("Promise.race");
 	});
 
-	test("drains in-flight dreamers (Promise.race with timeout)", () => {
-		expect(body).toContain("awaitInFlightDreamers");
+	test("drains the shutting-down session's recomp through withTimeout", () => {
+		expect(body).toMatch(
+			/withTimeout\(\s*awaitInFlightRecomps\(sessionId\),\s*SHUTDOWN_DRAIN_MS,?\s*\)/,
+		);
+	});
+
+	test("drains the current extension owner's dreamers through withTimeout", () => {
+		expect(body).toMatch(
+			/withTimeout\(\s*awaitInFlightDreamers\(dreamerRegistrationOwner\),\s*SHUTDOWN_DRAIN_MS,?\s*\)/,
+		);
+	});
+
+	test("stops Dreamer registration before draining its work", () => {
+		const shutdownAt = body.indexOf("sessionShuttingDown = true");
+		const unregisterAt = body.indexOf("unregisterPiDreamerProject");
+		const drainAt = body.indexOf("awaitInFlightDreamers");
+		expect(shutdownAt).toBeGreaterThanOrEqual(0);
+		expect(unregisterAt).toBeGreaterThanOrEqual(0);
+		expect(drainAt).toBeGreaterThanOrEqual(0);
+		expect(shutdownAt).toBeLessThan(unregisterAt);
+		expect(unregisterAt).toBeLessThan(drainAt);
+		expect(INDEX_SRC).toMatch(
+			/function syncDreamerProjectRegistration[\s\S]*?if \(sessionShuttingDown\) return;/,
+		);
 	});
 
 	test("drain timeout uses unref/clear helper", () => {

packages/pi-plugin/src/commands/ctx-commands.test.ts

@@ -29,8 +29,10 @@ interface AppendedEntry {
 interface MockCommandContext {
 	cwd: string;
 	hasUI?: boolean;
+	mode?: "rpc";
 	ui: {
 		custom: (factory: unknown, options?: unknown) => Promise<unknown>;
+		notify?: (text: string, type?: string) => void;
 		setStatus?: (key: string, text: string) => void;
 	};
 	model?: {
@@ -152,6 +154,34 @@ describe("Pi Magic Context commands", () => {
 		expect(sent[0]?.data.text).toContain("## Magic Status");
 	});
 
+	it("presents /ctx-status through the live RPC command context", async () => {
+		const db = createDb();
+		const { pi, handlers } = createMockPi();
+		const shownA: unknown[] = [];
+		const shownB: unknown[] = [];
+		const rpcCtx = (sessionId: string, shown: unknown[]) => ({
+			...createCtx(sessionId),
+			mode: "rpc" as const,
+			ui: {
+				async custom(factory: unknown) {
+					shown.push(factory);
+					return undefined;
+				},
+				notify() {},
+			},
+		});
+		registerCtxStatusCommand(pi as never, {
+			db,
+			projectIdentity: "/tmp/project",
+		});
+
+		await handlers.get("ctx-status")?.("", rpcCtx("ses-a", shownA));
+		await handlers.get("ctx-status")?.("", rpcCtx("ses-b", shownB));
+
+		expect(shownA).toHaveLength(1);
+		expect(shownB).toHaveLength(1);
+	});
+
 	it("/ctx-status keeps the persisted usable limit when command context omits maxTokens", async () => {
 		const db = createDb();
 		const sessionId = "ses-status-persisted-reserve";
@@ -259,17 +289,24 @@ describe("Pi Magic Context commands", () => {
 	it("registers /ctx-dream and starts a run (Dreamer v2 manual path)", async () => {
 		const db = createDb();
 		const { pi, handlers, sent } = createMockPi();
+		const registrationCwds: string[] = [];
 
 		registerCtxDreamCommand(pi as never, {
 			db,
 			projectDir: "/tmp/project",
 			projectIdentity: "/tmp/project",
+			registrationOwner: {},
+			ensureRegistered: (ctx) => {
+				registrationCwds.push(ctx.cwd);
+			},
 		});
 		// Not registered with the dreamer timer in this unit test, so runManual
 		// throws "not registered" → the handler reports the failure. We only
 		// assert the command is wired and emits a /ctx-dream status message.
+		// The injected registration sync runs immediately before runManual.
 		await handlers.get("ctx-dream")?.("", createCtx());
 
+		expect(registrationCwds).toEqual(["/tmp/project"]);
 		expect(sent[0]?.customType).toBe("ctx-status");
 		expect(sent[0]?.data.text).toContain("/ctx-dream");
 	});
@@ -282,6 +319,7 @@ describe("Pi Magic Context commands", () => {
 			db,
 			projectDir: "/tmp/project",
 			projectIdentity: "/tmp/project",
+			registrationOwner: {},
 		});
 
 		await handlers.get("ctx-dream")?.("verify", createCtx());
@@ -311,6 +349,7 @@ describe("Pi Magic Context commands", () => {
 			db,
 			projectDir: "/tmp/project",
 			projectIdentity: "/tmp/project",
+			registrationOwner: {},
 			dreamerEnabled: false,
 		});
 		await handlers.get("ctx-dream")?.("", createCtx());
@@ -326,6 +365,7 @@ describe("Pi Magic Context commands", () => {
 			db,
 			projectDir: "/tmp/project-a",
 			projectIdentity: "/tmp/project-a",
+			registrationOwner: {},
 			resolveProject: (ctx) => ({
 				projectDir: ctx.cwd,
 				projectIdentity: ctx.cwd,

packages/pi-plugin/src/commands/ctx-dream.ts

@@ -9,7 +9,7 @@ import {
 import type { ContextDatabase } from "@magic-context/core/features/magic-context/storage";
 import { sessionLog } from "@magic-context/core/shared/logger";
 import { runPiDreamForProject } from "../dreamer";
-import { sendCtxStatusMessage } from "./pi-command-utils";
+import { createCtxStatusSender } from "./pi-command-utils";
 
 export function registerCtxDreamCommand(
 	pi: ExtensionAPI,
@@ -24,11 +24,14 @@ export function registerCtxDreamCommand(
 		dreamerEnabled?: boolean;
 		resolveDreamerEnabled?: (ctx: { cwd: string }) => boolean | undefined;
 		onProjectSeen?: (projectIdentity: string) => void;
+		ensureRegistered?: (ctx: { cwd: string }) => void | Promise<void>;
+		registrationOwner: object;
 	},
 ): void {
 	pi.registerCommand("ctx-dream", {
 		description: "Run Magic Context dreamer tasks for this project now",
 		handler: async (args, ctx) => {
+			const sendStatus = createCtxStatusSender(pi, ctx);
 			const project = deps.resolveProject?.(ctx) ?? {
 				projectDir: deps.projectDir,
 				projectIdentity: deps.projectIdentity,
@@ -43,8 +46,7 @@ export function registerCtxDreamCommand(
 			let task: DreamTaskName | undefined;
 			if (requested) {
 				if (!isCanonicalDreamTask(requested)) {
-					sendCtxStatusMessage(
-						pi,
+					sendStatus(
 						{
 							title: "/ctx-dream",
 							text: `## /ctx-dream\n\nUnknown task "${requested}".`,
@@ -60,8 +62,7 @@ export function registerCtxDreamCommand(
 				task = requested;
 			}
 			if (dreamerEnabled === false) {
-				sendCtxStatusMessage(
-					pi,
+				sendStatus(
 					{
 						title: "/ctx-dream",
 						text: "## /ctx-dream\n\nDreamer is disabled for this project (`dreamer.disable=true`).",
@@ -83,8 +84,7 @@ export function registerCtxDreamCommand(
 
 			// Tell the user we're starting a real run, including the read-only count
 			// captured before the task acquires its lease.
-			sendCtxStatusMessage(
-				pi,
+			sendStatus(
 				{
 					title: "/ctx-dream",
 					text: [
@@ -108,9 +108,11 @@ export function registerCtxDreamCommand(
 
 			// Dreamer v2: run due/forced tasks now via the per-task scheduler.
 			try {
+				await deps.ensureRegistered?.(ctx);
 				const result = await runPiDreamForProject(
 					project.projectIdentity,
 					task,
+					deps.registrationOwner,
 				);
 				const lines: string[] = [];
 				if (result.ran.length > 0) lines.push(`Ran: ${result.ran.join(", ")}`);
@@ -140,12 +142,12 @@ export function registerCtxDreamCommand(
 				}
 				if (lines.length === 0) lines.push("No enabled dream tasks to run.");
 
-				sendCtxStatusMessage(
-					pi,
+				sendStatus(
 					{
 						title: "/ctx-dream",
 						text: ["## /ctx-dream", "", ...lines].join("\n"),
 						level: result.ran.length > 0 ? "success" : "info",
+						rpcDisplay: "dialog",
 					},
 					{
 						projectDir: project.projectDir,
@@ -155,8 +157,7 @@ export function registerCtxDreamCommand(
 			} catch (error) {
 				const message = error instanceof Error ? error.message : String(error);
 				sessionLog(project.projectIdentity, `/ctx-dream failed: ${message}`);
-				sendCtxStatusMessage(
-					pi,
+				sendStatus(
 					{
 						title: "/ctx-dream",
 						text: [

packages/pi-plugin/src/commands/ctx-embed.ts

@@ -12,7 +12,7 @@ import {
 } from "@magic-context/core/hooks/magic-context/embed-session-state";
 import { formatEmbedStatusText } from "@magic-context/core/hooks/magic-context/format-embed-status";
 import { ensureProjectRegisteredFromPiDirectory } from "../embedding-bootstrap";
-import { resolveSessionId, sendCtxStatusMessage } from "./pi-command-utils";
+import { createCtxStatusSender, resolveSessionId } from "./pi-command-utils";
 
 const EMBED_PROGRESS_COMPARTMENT_STEP = 8;
 const EMBED_PROGRESS_MIN_INTERVAL_MS = 10_000;
@@ -159,9 +159,10 @@ export function registerCtxEmbedCommand(
 		description:
 			"Embedding status, or start/pause history compartment embedding (start | pause)",
 		handler: async (args, ctx) => {
+			const sendStatus = createCtxStatusSender(pi, ctx);
 			const sessionId = resolveSessionId(ctx);
 			if (!sessionId) {
-				sendCtxStatusMessage(pi, {
+				sendStatus({
 					title: "/ctx-embed",
 					text: "## /ctx-embed\n\nNo active Pi session is available.",
 					level: "error",
@@ -185,7 +186,7 @@ export function registerCtxEmbedCommand(
 					project.projectIdentity,
 					sessionId,
 				);
-				sendCtxStatusMessage(pi, {
+				sendStatus({
 					title: "/ctx-embed",
 					text: `## /ctx-embed\n\nPaused at ${cov.session.embedded}/${cov.session.total} compartments embedded.`,
 					level: "info",
@@ -194,7 +195,7 @@ export function registerCtxEmbedCommand(
 			}
 
 			if (memoryEnabled === false) {
-				sendCtxStatusMessage(pi, {
+				sendStatus({
 					title: "/ctx-embed",
 					text: "## /ctx-embed\n\nMemory is disabled for this project, so there is no semantic embedding to backfill.",
 					level: "info",
@@ -211,18 +212,18 @@ export function registerCtxEmbedCommand(
 					sessionId,
 					{
 						onStatus: (status) =>
-							sendCtxStatusMessage(pi, {
+							sendStatus({
 								title: "/ctx-embed",
 								...status,
 							}),
 					},
 				);
-				sendCtxStatusMessage(pi, { title: "/ctx-embed", text, level });
+				sendStatus({ title: "/ctx-embed", text, level });
 				return;
 			}
 
 			if (sub !== "") {
-				sendCtxStatusMessage(pi, {
+				sendStatus({
 					title: "/ctx-embed",
 					text: "## /ctx-embed\n\nUsage: `/ctx-embed` (status), `/ctx-embed start`, or `/ctx-embed pause`.",
 					level: "info",
@@ -236,10 +237,11 @@ export function registerCtxEmbedCommand(
 				sessionId,
 			);
 			const statusText = formatEmbedStatusText(coverage, { status: "idle" });
-			sendCtxStatusMessage(pi, {
+			sendStatus({
 				title: "/ctx-embed",
 				text: `## Embedding Status\n\n${statusText}`,
 				level: "info",
+				rpcDisplay: "dialog",
 			});
 		},
 	});