From a99696bd09e437fa7d475ebcb80ad9a0364fbb3a Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Thu, 25 Jun 2026 00:20:19 +0200 Subject: [PATCH] fix: persist tracked subsession links --- .changeset/persist-subsession-links.md | 5 + src/server/sessions/piSessionService.test.ts | 371 ++++++++++++++++++- src/server/sessions/piSessionService.ts | 325 +++++++++++++++- 3 files changed, 692 insertions(+), 9 deletions(-) create mode 100644 .changeset/persist-subsession-links.md diff --git a/.changeset/persist-subsession-links.md b/.changeset/persist-subsession-links.md new file mode 100644 index 0000000..8199eaa --- /dev/null +++ b/.changeset/persist-subsession-links.md @@ -0,0 +1,5 @@ +--- +"@jmfederico/pi-web": patch +--- + +Persist tracked subsession links in session history so parents can list, check, and read child sessions after the session daemon restarts, and reopened children can resume parent notifications. diff --git a/src/server/sessions/piSessionService.test.ts b/src/server/sessions/piSessionService.test.ts index 9de755f..543105a 100644 --- a/src/server/sessions/piSessionService.test.ts +++ b/src/server/sessions/piSessionService.test.ts @@ -1,3 +1,6 @@ +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { AuthStorage, ModelRegistry } from "@earendil-works/pi-coding-agent"; import { describe, expect, it, vi } from "vitest"; import type { GlobalSessionEvent, SessionUiEvent } from "../../shared/apiTypes.js"; @@ -32,11 +35,12 @@ interface TestSession extends PiAgentSession { getFollowUpMessages: () => readonly string[]; } -function fakeSessionManager(cwd = "/workspace"): PiSessionManager { +function fakeSessionManager(cwd = "/workspace", patch: Partial = {}): PiSessionManager { return { getCwd: () => cwd, getBranch: () => [], getLeafId: () => "leaf-1", + ...patch, }; } @@ -141,6 +145,16 @@ function sessionGateway(records: ReturnType[]): SessionGat }; } +function emptyArchiveStore(): NonNullable { + return { + list: () => Promise.resolve([]), + get: () => Promise.resolve(undefined), + archive: () => Promise.reject(new Error("archive should not be called")), + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + }; +} + describe("PiSessionService", () => { it("starts sessions through an injected runtime creator", async () => { const hub = new CapturingSessionEventHub(); @@ -887,6 +901,361 @@ describe("PiSessionService", () => { await service.dispose(); }); + it("persists tracked child links in the parent and child sessions", async () => { + const parentPersisted: { customType: string; data?: unknown }[] = []; + const childPersisted: { customType: string; data?: unknown }[] = []; + const parent = fakeRuntime("parent-1", { + sessionFile: "/tmp/parent-1.jsonl", + sessionManager: fakeSessionManager("/workspace", { + appendCustomEntry: (customType, data) => { + parentPersisted.push({ customType, data }); + return "parent-entry-1"; + }, + }), + }); + const child = fakeRuntime("child-1", { + sessionFile: "/tmp/child-1.jsonl", + sessionManager: fakeSessionManager("/workspace-feature", { + appendCustomEntry: (customType, data) => { + childPersisted.push({ customType, data }); + return "child-entry-1"; + }, + }), + }); + const runtimes = [parent.runtime, child.runtime]; + let index = 0; + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? child.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: sessionGateway([]), + archiveStore: emptyArchiveStore(), + spawnTargets: { resolveSpawnTarget: () => Promise.resolve({ allowed: true, cwd: "/workspace-feature" }) }, + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + await service.spawnSubsession({ spawningCwd: "/workspace", parentSessionId: "parent-1", parentSessionFile: "/tmp/parent-1.jsonl", prompt: "do the slice", cwd: "/workspace-feature" }); + + expect(parentPersisted).toEqual([ + { + customType: "pi-web.subsession.link", + data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: "/tmp/child-1.jsonl", cwd: "/workspace-feature" }, + }, + ]); + expect(childPersisted).toEqual([ + { + customType: "pi-web.subsession.spawned", + data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" }, + }, + ]); + await service.dispose(); + }); + + it("hydrates persisted child links after a service restart so the parent can inspect them", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-")); + const parentFile = join(tempDir, "parent.jsonl"); + const childFile = join(tempDir, "child.jsonl"); + await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8"); + await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8"); + + try { + const childManager = fakeSessionManager("/workspace-feature", { + getBranch: () => [{ type: "message", message: { role: "assistant", content: "finished" } }], + }); + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace-feature" } }], + }), + }); + const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const runtimes = [parent.runtime, child.runtime]; + let index = 0; + const open = vi.fn(() => childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? child.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.checkSubsession("parent-1", "child-1")).resolves.toEqual({ + sessionId: "child-1", + cwd: "/workspace-feature", + status: "idle", + finalText: "finished", + messageCount: 1, + }); + expect(open).toHaveBeenCalledWith(childFile); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("ignores stale persisted child links when the child no longer records the parent", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-stale-")); + const parentFile = join(tempDir, "parent.jsonl"); + const childFile = join(tempDir, "child.jsonl"); + await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8"); + await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature" })}\n`, "utf8"); + + try { + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace-feature" } }], + }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("hydrates persisted links to archived children without scanning unrelated child headers", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: "/sessions/child-1.jsonl", cwd: "/workspace-feature" } }], + }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: { + ...emptyArchiveStore(), + list: () => Promise.resolve([]), + get: (sessionId) => Promise.resolve(sessionId === "child-1" ? { sessionId: "child-1", cwd: "/workspace-feature", archivedAt: "2026-01-01T00:00:00.000Z", parentSessionPath: parentFile } : undefined), + isArchived: (sessionId) => Promise.resolve(sessionId === "child-1"), + }, + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([ + { sessionId: "child-1", cwd: "/workspace-feature", status: "archived" }, + ]); + await service.dispose(); + }); + + it("does not hydrate parent links without a child file or exact archived child validation", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child", cwd: "/workspace-feature" } }], + }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: { + ...emptyArchiveStore(), + get: (sessionId) => Promise.resolve(sessionId === "child" ? { sessionId: "child-fork", cwd: "/workspace-feature", archivedAt: "2026-01-01T00:00:00.000Z", parentSessionPath: parentFile } : undefined), + }, + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + }); + + it("does not invent subsession links from existing child session headers", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const childRecord = { ...sessionRecord("child-1", "/workspace-feature"), path: "/sessions/child-1.jsonl", parentSessionPath: parentFile }; + const parent = fakeRuntime("parent-1", { + sessionFile: parentFile, + sessionManager: fakeSessionManager("/workspace", { getEntries: () => [] }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(parent.runtime), + sessionManager: { create: () => parent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([childRecord]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + }); + + it("does not hydrate copied parent links when the opened parent has a different id", async () => { + const forkedParent = fakeRuntime("parent-fork-1", { + sessionFile: "/sessions/parent-fork-1.jsonl", + sessionManager: fakeSessionManager("/workspace", { + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: "/sessions/child-1.jsonl", cwd: "/workspace-feature" } }], + }), + }); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(forkedParent.runtime), + sessionManager: { create: () => forkedParent.session.sessionManager, list: () => Promise.resolve([]), listAll: () => Promise.resolve([]), open: () => fakeSessionManager() }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.start("/workspace"); + + await expect(service.listSubsessions("parent-fork-1")).resolves.toEqual([]); + await service.dispose(); + }); + + it("relinks a spawned child when the child session is opened after restart", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-open-child-")); + const parentFile = join(tempDir, "parent.jsonl"); + const childFile = join(tempDir, "child.jsonl"); + await writeFile(parentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8"); + await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8"); + + try { + const childManager = fakeSessionManager("/workspace-feature", { + getHeader: () => ({ parentSession: parentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + const parentManager = fakeSessionManager("/workspace"); + const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager }); + const runtimes = [child.runtime, parent.runtime]; + let index = 0; + const open = vi.fn((path: string) => path === parentFile ? parentManager : childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? parent.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...sessionRecord("child-1", "/workspace-feature"), path: childFile, parentSessionPath: parentFile }]), + listAll: () => Promise.resolve([]), + open, + }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("child-1", "/workspace-feature")); + child.session.isStreaming = true; + child.emit({ type: "agent_start" }); + child.session.isStreaming = false; + child.emit({ type: "agent_end" }); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(parent.calls.sendCustomMessage).toHaveLength(1); + expect(parent.calls.sendCustomMessage[0]?.message.content).toContain("Subsession child-1 stopped working"); + expect(open).toHaveBeenCalledWith(parentFile); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("does not relink a child marker when the child header points at a different parent id", async () => { + const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-wrong-parent-")); + const mismatchedParentFile = join(tempDir, "other-parent.jsonl"); + const actualParentFile = join(tempDir, "parent.jsonl"); + const childFile = join(tempDir, "child.jsonl"); + await writeFile(mismatchedParentFile, `${JSON.stringify({ type: "session", version: 3, id: "other-parent", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace" })}\n`, "utf8"); + await writeFile(childFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: mismatchedParentFile })}\n`, "utf8"); + + try { + const childManager = fakeSessionManager("/workspace-feature", { + getHeader: () => ({ parentSession: mismatchedParentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + const parent = fakeRuntime("parent-1", { sessionFile: actualParentFile, sessionManager: fakeSessionManager("/workspace") }); + const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager }); + const runtimes = [child.runtime, parent.runtime]; + let index = 0; + const open = vi.fn((path: string) => path === actualParentFile ? parent.session.sessionManager : childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: () => { + const runtime = runtimes[index] ?? parent.runtime; + index += 1; + return Promise.resolve(runtime); + }, + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...sessionRecord("child-1", "/workspace-feature"), path: childFile, parentSessionPath: mismatchedParentFile }]), + listAll: () => Promise.resolve([{ ...sessionRecord("parent-1", "/workspace"), path: actualParentFile }]), + open, + }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("child-1", "/workspace-feature")); + child.session.isStreaming = true; + child.emit({ type: "agent_start" }); + child.session.isStreaming = false; + child.emit({ type: "agent_end" }); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(parent.calls.sendCustomMessage).toHaveLength(0); + expect(open).not.toHaveBeenCalledWith(actualParentFile); + await service.dispose(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + }); + + it("does not relink copied child markers when the opened child has a different id", async () => { + const parentFile = "/sessions/parent-1.jsonl"; + const childFile = "/sessions/child-fork-1.jsonl"; + const childManager = fakeSessionManager("/workspace-feature", { + getHeader: () => ({ parentSession: parentFile }), + getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }], + }); + const child = fakeRuntime("child-fork-1", { sessionFile: childFile, sessionManager: childManager }); + const open = vi.fn(() => childManager); + const service = new PiSessionService(new CapturingSessionEventHub(), { + createAgentRuntime: runtimeCreator(child.runtime), + sessionManager: { + create: () => childManager, + list: () => Promise.resolve([{ ...sessionRecord("child-fork-1", "/workspace-feature"), path: childFile, parentSessionPath: parentFile }]), + listAll: () => Promise.resolve([]), + open, + }, + archiveStore: emptyArchiveStore(), + heartbeatIntervalMs: 60_000, + }); + + await service.status(sessionRef("child-fork-1", "/workspace-feature")); + child.session.isStreaming = true; + child.emit({ type: "agent_start" }); + child.session.isStreaming = false; + child.emit({ type: "agent_end" }); + await new Promise((resolve) => setTimeout(resolve, 20)); + + expect(open).not.toHaveBeenCalledWith(parentFile); + await expect(service.listSubsessions("parent-1")).resolves.toEqual([]); + await service.dispose(); + }); + it("notifies the parent once when the tracked child stops working", async () => { const { parent, child, service } = subsessionService({ allowed: true, cwd: "/workspace-feature" }); await service.start("/workspace"); diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index 3562925..b414534 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -1,4 +1,4 @@ -import { readFile, writeFile } from "node:fs/promises"; +import { open, readFile, writeFile } from "node:fs/promises"; import type { Api, ImageContent, Model } from "@earendil-works/pi-ai"; import { AuthStorage, @@ -81,6 +81,26 @@ interface QueuedPrompt { echoUserMessage?: boolean; } +interface TrackedSubsessionLink { + parentSessionId: string; + childSessionId: string; + childSessionFile?: string; + parentSessionFile?: string; + cwd?: string; +} + +interface PersistedParentSubsessionLink { + spawnedBySessionId: string; + spawnedSessionId: string; + spawnedSessionFile?: string; + cwd?: string; +} + +interface PersistedChildSubsessionLink { + spawnedBySessionId: string; + spawnedSessionId: string; +} + function requirePromptText(value: unknown): string { if (typeof value !== "string") throw new Error("Prompt text is required"); return value; @@ -123,8 +143,10 @@ type ModelRegistryInstance = ReturnType; export interface PiSessionManager { getCwd(): string; getBranch(): unknown[]; + getEntries?(): readonly unknown[]; getLeafId(): string | null; getHeader?(): { parentSession?: string } | null | undefined; + appendCustomEntry?(customType: string, data?: unknown): string; } export interface PiSessionManagerGateway { @@ -290,6 +312,10 @@ export class PiSessionService { private readonly subsessionParents = new Map(); /** Parent session id -> the set of tracked subsession ids it spawned. */ private readonly subsessionChildren = new Map>(); + /** Tracked subsession id -> persisted recovery details for the child. */ + private readonly subsessionLinks = new Map(); + /** Parent session ids whose persisted links have already been loaded. */ + private readonly subsessionHydratedParents = new Set(); /** * Tracked subsession id -> whether a completion notification is armed. * Armed when the child starts working; firing on completion disarms it so a @@ -362,6 +388,8 @@ export class PiSessionService { this.authLossWarnings.clear(); this.subsessionParents.clear(); this.subsessionChildren.clear(); + this.subsessionLinks.clear(); + this.subsessionHydratedParents.clear(); this.subsessionNotifyArmed.clear(); await Promise.all(activeSessions.map(async (active) => { active.unsubscribe(); @@ -439,7 +467,16 @@ export class PiSessionService { const decision = await this.spawnTargets.resolveSpawnTarget(input.spawningCwd, input.cwd); if (!decision.allowed) throw spawnTargetError(decision); const created = await this.start(decision.cwd, input.parentSessionFile); - this.registerSubsession(input.parentSessionId, created.id); + const parentSessionFile = nonEmptyString(input.parentSessionFile); + const link = { + childSessionId: created.id, + ...(created.path === "" ? {} : { childSessionFile: created.path }), + ...(parentSessionFile === undefined ? {} : { parentSessionFile }), + cwd: decision.cwd, + }; + this.registerSubsession(input.parentSessionId, link); + this.persistSubsessionLink(input.parentSessionId, link); + this.persistSubsessionChildMarker(input.parentSessionId, created.id); await this.prompt(created.id, input.prompt); this.logger.info( { parentSessionId: input.parentSessionId, sessionId: created.id, cwd: decision.cwd, promptLength: input.prompt.length }, @@ -450,6 +487,7 @@ export class PiSessionService { /** Summaries of the tracked subsessions spawned by `parentSessionId`. */ async listSubsessions(parentSessionId: string): Promise { + await this.hydrateSubsessionsForParent(parentSessionId); const childIds = this.subsessionChildren.get(parentSessionId); if (childIds === undefined) return []; return Promise.all([...childIds].map(async (childId) => ({ sessionId: childId, ...(await this.subsessionSummaryFields(childId)) }))); @@ -482,18 +520,144 @@ export class PiSessionService { /** Open a session after verifying it is one of the caller's tracked children. */ private async openSubsession(parentSessionId: string, sessionId: string): Promise { + await this.hydrateSubsessionsForParent(parentSessionId); if (this.subsessionParents.get(sessionId) !== parentSessionId) { throw new Error(`Session ${sessionId} is not one of your subsessions`); } - return this.getOrOpen(sessionId); + return this.getOrOpenTrackedSubsession(sessionId); } - private registerSubsession(parentSessionId: string, childSessionId: string): void { + private registerSubsession(parentSessionId: string, link: Omit): void { + const childSessionId = link.childSessionId; + const previousParentId = this.subsessionParents.get(childSessionId); + if (previousParentId !== undefined && previousParentId !== parentSessionId) { + const previousChildren = this.subsessionChildren.get(previousParentId); + previousChildren?.delete(childSessionId); + if (previousChildren?.size === 0) this.subsessionChildren.delete(previousParentId); + } + this.subsessionParents.set(childSessionId, parentSessionId); const children = this.subsessionChildren.get(parentSessionId) ?? new Set(); children.add(childSessionId); this.subsessionChildren.set(parentSessionId, children); - this.subsessionNotifyArmed.set(childSessionId, false); + + const previous = this.subsessionLinks.get(childSessionId); + this.subsessionLinks.set(childSessionId, mergeSubsessionLink(previous, { ...link, parentSessionId })); + if (!this.subsessionNotifyArmed.has(childSessionId)) this.subsessionNotifyArmed.set(childSessionId, false); + } + + private unregisterSubsession(childSessionId: string): void { + const parentSessionId = this.subsessionParents.get(childSessionId); + this.subsessionParents.delete(childSessionId); + this.subsessionLinks.delete(childSessionId); + this.subsessionNotifyArmed.delete(childSessionId); + if (parentSessionId === undefined) return; + const children = this.subsessionChildren.get(parentSessionId); + children?.delete(childSessionId); + if (children?.size === 0) this.subsessionChildren.delete(parentSessionId); + } + + private persistSubsessionLink(parentSessionId: string, link: Omit): void { + const parent = this.active.get(parentSessionId)?.runtime.session; + if (parent === undefined) return; + if (parent.sessionManager.appendCustomEntry === undefined) return; + try { + parent.sessionManager.appendCustomEntry(SUBSESSION_LINK_CUSTOM_TYPE, persistedParentSubsessionLinkData(parentSessionId, link)); + } catch (error: unknown) { + this.logger.info( + { parentSessionId, sessionId: link.childSessionId, error: error instanceof Error ? error.message : String(error) }, + "failed to persist subsession link", + ); + } + } + + private persistSubsessionChildMarker(parentSessionId: string, childSessionId: string): void { + const child = this.active.get(childSessionId)?.runtime.session; + if (child === undefined) return; + if (child.sessionManager.appendCustomEntry === undefined) return; + try { + child.sessionManager.appendCustomEntry(SUBSESSION_CHILD_LINK_CUSTOM_TYPE, persistedChildSubsessionLinkData(parentSessionId, childSessionId)); + } catch (error: unknown) { + this.logger.info( + { parentSessionId, sessionId: childSessionId, error: error instanceof Error ? error.message : String(error) }, + "failed to persist subsession child marker", + ); + } + } + + private async hydrateSubsessionsForParent(parentSessionId: string): Promise { + if (this.subsessionHydratedParents.has(parentSessionId)) return; + const parent = this.active.get(parentSessionId)?.runtime.session; + if (parent === undefined) return; + + const parentSessionFile = nonEmptyString(parent.sessionFile); + await this.registerPersistedSubsessionLinks(parentSessionId, parent, parentSessionFile); + this.subsessionHydratedParents.add(parentSessionId); + } + + private async registerPersistedSubsessionLinks(parentSessionId: string, parent: PiAgentSession, parentSessionFile: string | undefined): Promise { + const entries = parent.sessionManager.getEntries?.() ?? parent.sessionManager.getBranch(); + for (const entry of entries) { + const link = parsePersistedParentSubsessionLink(entry); + if (link === undefined) continue; + if (link.spawnedBySessionId !== parentSessionId) continue; + if (!await this.persistedSubsessionLinkMatchesParent(parentSessionFile, link)) continue; + this.registerSubsession(parentSessionId, trackedSubsessionLinkFromParentLink(link, parentSessionFile)); + } + } + + private async persistedSubsessionLinkMatchesParent(parentSessionFile: string | undefined, link: PersistedParentSubsessionLink): Promise { + if (parentSessionFile === undefined) return false; + if (link.spawnedSessionFile !== undefined) { + const header = await readSessionHeaderSummary(link.spawnedSessionFile); + if (header?.id === link.spawnedSessionId) { + return header.parentSession !== undefined && sessionPathsEqual(header.parentSession, parentSessionFile); + } + } + + const archived = await this.getArchivedExact(link.spawnedSessionId); + return archived?.parentSessionPath !== undefined && sessionPathsEqual(archived.parentSessionPath, parentSessionFile); + } + + private async recoverSubsessionTrackingForOpenedSession(session: PiAgentSession): Promise { + const entries = session.sessionManager.getEntries?.() ?? session.sessionManager.getBranch(); + let marker: PersistedChildSubsessionLink | undefined; + for (const entry of entries) { + const parsed = parsePersistedChildSubsessionLink(entry); + if (parsed?.spawnedSessionId === session.sessionId) marker = parsed; + } + if (marker === undefined) return; + + const parentSessionFile = await parentSessionFileForSession(session); + if (parentSessionFile === undefined) return; + const parentHeader = await readSessionHeaderSummary(parentSessionFile); + if (parentHeader?.id !== marker.spawnedBySessionId) return; + const childSessionFile = nonEmptyString(session.sessionFile); + this.registerSubsession(marker.spawnedBySessionId, { + childSessionId: session.sessionId, + ...(childSessionFile === undefined ? {} : { childSessionFile }), + parentSessionFile, + cwd: session.sessionManager.getCwd(), + }); + } + + private async getOrOpenTrackedSubsession(sessionId: string): Promise { + const active = this.active.get(sessionId); + if (active !== undefined) return active.runtime.session; + + const archived = await this.getArchivedExact(sessionId); + if (archived?.archivePath !== undefined) return (await this.create(this.sessionManager.open(archived.archivePath), archived.cwd)).runtime.session; + + const link = this.subsessionLinks.get(sessionId); + if (link?.childSessionFile !== undefined) { + const header = await readSessionHeaderSummary(link.childSessionFile); + if (header?.id === sessionId) { + const sessionManager = this.sessionManager.open(link.childSessionFile); + return (await this.create(sessionManager, link.cwd ?? sessionManager.getCwd())).runtime.session; + } + } + + return this.getOrOpen(sessionId); } private async subsessionSummaryFields(childSessionId: string): Promise<{ cwd: string; status: SubsessionStatus }> { @@ -501,13 +665,18 @@ export class PiSessionService { if (active !== undefined) { return { cwd: active.runtime.cwd, status: await this.subsessionStatus(active.runtime.session) }; } - const archived = await this.archiveStore.get(childSessionId); + const archived = await this.getArchivedExact(childSessionId); if (archived !== undefined) return { cwd: archived.cwd, status: "archived" }; + const link = this.subsessionLinks.get(childSessionId); + if (link?.childSessionFile !== undefined && (await readSessionHeaderSummary(link.childSessionFile))?.id === childSessionId) { + return { cwd: link.cwd ?? "", status: "idle" }; + } + if (link?.cwd !== undefined) return { cwd: link.cwd, status: "unknown" }; return { cwd: "", status: "unknown" }; } private async subsessionStatus(session: PiAgentSession): Promise { - if (await this.archiveStore.isArchived(session.sessionId)) return "archived"; + if (await this.getArchivedExact(session.sessionId) !== undefined) return "archived"; if (this.hasActiveWork(session)) return "working"; if (this.activities.get(session.sessionId)?.phase === "error") return "error"; return "idle"; @@ -536,6 +705,19 @@ export class PiSessionService { void this.notifyParentOfSubsession(parentId, childId, text); } + private async getOrOpenParentForSubsession(parentSessionId: string, childSessionId: string): Promise { + const active = this.activeForLookup(parentSessionId); + if (active !== undefined) return active.runtime.session; + + const parentSessionFile = this.subsessionLinks.get(childSessionId)?.parentSessionFile; + if (parentSessionFile !== undefined && (await readSessionHeaderSummary(parentSessionFile))?.id === parentSessionId) { + const sessionManager = this.sessionManager.open(parentSessionFile); + return (await this.create(sessionManager, sessionManager.getCwd())).runtime.session; + } + + return this.getOrOpen(parentSessionId); + } + /** * Deliver a subsession-completion notice to the parent as a system-authored * custom message rather than a user message, so it is not attributed to the @@ -545,7 +727,7 @@ export class PiSessionService { */ private async notifyParentOfSubsession(parentId: string, childId: string, text: string): Promise { try { - const session = await this.getOrOpen(parentId); + const session = await this.getOrOpenParentForSubsession(parentId, childId); await session.sendCustomMessage( { customType: SUBSESSION_NOTIFICATION_CUSTOM_TYPE, content: text, display: true, details: { sessionId: childId } }, { triggerTurn: true, deliverAs: "followUp" }, @@ -809,6 +991,8 @@ export class PiSessionService { const sessionFile = session.sessionFile; if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); await clearParentSession(sessionFile); + clearParentSessionHeader(session.sessionManager); + this.unregisterSubsession(session.sessionId); } async abort(ref: PiSessionLookup): Promise { @@ -961,6 +1145,11 @@ export class PiSessionService { return archived; } + private async getArchivedExact(sessionId: string): Promise { + const archived = await this.archiveStore.get(sessionId); + return archived?.sessionId === sessionId ? archived : undefined; + } + private activeForLookup(ref: PiSessionLookup): ActiveSession | undefined { const sessionId = sessionIdFromLookup(ref); const exact = this.active.get(sessionId); @@ -979,8 +1168,10 @@ export class PiSessionService { runtime.setRebindSession(async (session) => { await this.bindSessionExtensions(session); this.bindRuntime(active); + await this.recoverSubsessionTrackingForOpenedSession(session); }); this.active.set(runtime.session.sessionId, active); + await this.recoverSubsessionTrackingForOpenedSession(runtime.session); this.publishStatus(runtime.session); return active; } @@ -1411,6 +1602,113 @@ function isDefined(value: T | undefined): value is T { return value !== undefined; } +function mergeSubsessionLink(previous: TrackedSubsessionLink | undefined, next: TrackedSubsessionLink): TrackedSubsessionLink { + return { + parentSessionId: next.parentSessionId, + childSessionId: next.childSessionId, + ...(previous?.childSessionFile === undefined ? {} : { childSessionFile: previous.childSessionFile }), + ...(previous?.parentSessionFile === undefined ? {} : { parentSessionFile: previous.parentSessionFile }), + ...(previous?.cwd === undefined ? {} : { cwd: previous.cwd }), + ...(next.childSessionFile === undefined ? {} : { childSessionFile: next.childSessionFile }), + ...(next.parentSessionFile === undefined ? {} : { parentSessionFile: next.parentSessionFile }), + ...(next.cwd === undefined ? {} : { cwd: next.cwd }), + }; +} + +function trackedSubsessionLinkFromParentLink(link: PersistedParentSubsessionLink, parentSessionFile: string | undefined): Omit { + return { + childSessionId: link.spawnedSessionId, + ...(link.spawnedSessionFile === undefined ? {} : { childSessionFile: link.spawnedSessionFile }), + ...(parentSessionFile === undefined ? {} : { parentSessionFile }), + ...(link.cwd === undefined ? {} : { cwd: link.cwd }), + }; +} + +function persistedParentSubsessionLinkData(parentSessionId: string, link: Omit): Record { + return { + version: 1, + spawnedBySessionId: parentSessionId, + spawnedSessionId: link.childSessionId, + ...(link.childSessionFile === undefined ? {} : { spawnedSessionFile: link.childSessionFile }), + ...(link.cwd === undefined ? {} : { cwd: link.cwd }), + }; +} + +function persistedChildSubsessionLinkData(parentSessionId: string, childSessionId: string): Record { + return { + version: 1, + spawnedBySessionId: parentSessionId, + spawnedSessionId: childSessionId, + }; +} + +function parsePersistedParentSubsessionLink(entry: unknown): PersistedParentSubsessionLink | undefined { + if (!isRecord(entry) || entry["type"] !== "custom" || entry["customType"] !== SUBSESSION_LINK_CUSTOM_TYPE) return undefined; + const data = entry["data"]; + if (!isRecord(data)) return undefined; + const spawnedBySessionId = getString(data, "spawnedBySessionId"); + const spawnedSessionId = getString(data, "spawnedSessionId"); + if (spawnedBySessionId === undefined || spawnedBySessionId === "" || spawnedSessionId === undefined || spawnedSessionId === "") return undefined; + const spawnedSessionFile = getString(data, "spawnedSessionFile"); + const cwd = getString(data, "cwd"); + return { + spawnedBySessionId, + spawnedSessionId, + ...(spawnedSessionFile === undefined || spawnedSessionFile === "" ? {} : { spawnedSessionFile }), + ...(cwd === undefined || cwd === "" ? {} : { cwd }), + }; +} + +function parsePersistedChildSubsessionLink(entry: unknown): PersistedChildSubsessionLink | undefined { + if (!isRecord(entry) || entry["type"] !== "custom" || entry["customType"] !== SUBSESSION_CHILD_LINK_CUSTOM_TYPE) return undefined; + const data = entry["data"]; + if (!isRecord(data)) return undefined; + const spawnedBySessionId = getString(data, "spawnedBySessionId"); + const spawnedSessionId = getString(data, "spawnedSessionId"); + if (spawnedBySessionId === undefined || spawnedBySessionId === "" || spawnedSessionId === undefined || spawnedSessionId === "") return undefined; + return { spawnedBySessionId, spawnedSessionId }; +} + +function nonEmptyString(value: string | undefined): string | undefined { + return value === undefined || value === "" ? undefined : value; +} + +function sessionPathsEqual(a: string, b: string): boolean { + return cwdPathsEqual(a, b); +} + +interface SessionHeaderSummary { + id: string; + parentSession?: string; +} + +async function readSessionHeaderSummary(sessionFile: string): Promise { + let file: Awaited> | undefined; + try { + file = await open(sessionFile, "r"); + const buffer = Buffer.alloc(4096); + const { bytesRead } = await file.read(buffer, 0, buffer.length, 0); + const firstLine = buffer.toString("utf8", 0, bytesRead).split("\n", 1)[0]; + if (firstLine === undefined || firstLine === "") return undefined; + const header: unknown = JSON.parse(firstLine); + if (!isRecord(header) || header["type"] !== "session" || typeof header["id"] !== "string") return undefined; + const parentSession = getString(header, "parentSession"); + return { id: header["id"], ...(parentSession === undefined ? {} : { parentSession }) }; + } catch { + return undefined; + } finally { + await file?.close().catch(() => undefined); + } +} + +async function parentSessionFileForSession(session: PiAgentSession): Promise { + const headerParentSession = nonEmptyString(session.sessionManager.getHeader?.()?.parentSession); + if (headerParentSession !== undefined) return headerParentSession; + const sessionFile = nonEmptyString(session.sessionFile); + if (sessionFile === undefined) return undefined; + return (await readSessionHeaderSummary(sessionFile))?.parentSession; +} + async function clearParentSession(sessionFile: string): Promise { const content = await readFile(sessionFile, "utf8"); const newlineIndex = content.indexOf("\n"); @@ -1423,6 +1721,11 @@ async function clearParentSession(sessionFile: string): Promise { await writeFile(sessionFile, `${JSON.stringify(header)}${rest}`, "utf8"); } +function clearParentSessionHeader(sessionManager: PiSessionManager): void { + const header = sessionManager.getHeader?.(); + if (header !== undefined && header !== null) delete header.parentSession; +} + function clearSessionQueue(session: PiAgentSession): void { session.clearQueue(); } @@ -1475,6 +1778,12 @@ function historyMessages(session: PiAgentSession): unknown[] { return messages; } +/** custom entry type used to persist parent -> child subsession links outside LLM context. */ +const SUBSESSION_LINK_CUSTOM_TYPE = "pi-web.subsession.link"; + +/** custom entry type used to mark a child as created by spawn_subsession. */ +const SUBSESSION_CHILD_LINK_CUSTOM_TYPE = "pi-web.subsession.spawned"; + /** customType marking a parent-facing subsession-completion notice. */ const SUBSESSION_NOTIFICATION_CUSTOM_TYPE = "subsession.completion";