fix: persist tracked subsession links

This commit is contained in:
Federico Jaramillo Martinez
2026-06-25 00:20:19 +02:00
parent deb4f9feca
commit a99696bd09
3 changed files with 692 additions and 9 deletions
+5
View File
@@ -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.
+370 -1
View File
@@ -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> = {}): PiSessionManager {
return {
getCwd: () => cwd,
getBranch: () => [],
getLeafId: () => "leaf-1",
...patch,
};
}
@@ -141,6 +145,16 @@ function sessionGateway(records: ReturnType<typeof sessionRecord>[]): SessionGat
};
}
function emptyArchiveStore(): NonNullable<PiSessionServiceDependencies["archiveStore"]> {
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");
+317 -8
View File
@@ -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<typeof ModelRegistry.create>;
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<string, string>();
/** Parent session id -> the set of tracked subsession ids it spawned. */
private readonly subsessionChildren = new Map<string, Set<string>>();
/** Tracked subsession id -> persisted recovery details for the child. */
private readonly subsessionLinks = new Map<string, TrackedSubsessionLink>();
/** Parent session ids whose persisted links have already been loaded. */
private readonly subsessionHydratedParents = new Set<string>();
/**
* 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<SubsessionSummary[]> {
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<PiAgentSession> {
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<TrackedSubsessionLink, "parentSessionId">): 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<string>();
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<TrackedSubsessionLink, "parentSessionId">): 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<void> {
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<void> {
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<boolean> {
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<void> {
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<PiAgentSession> {
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<SubsessionStatus> {
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<PiAgentSession> {
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<void> {
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<void> {
@@ -961,6 +1145,11 @@ export class PiSessionService {
return archived;
}
private async getArchivedExact(sessionId: string): Promise<ArchivedSessionRecord | undefined> {
const archived = await this.archiveStore.get(sessionId);
return archived?.sessionId === sessionId ? archived : undefined;
}
private activeForLookup(ref: PiSessionLookup): ActiveSession<PiSessionRuntime> | 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<T>(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<TrackedSubsessionLink, "parentSessionId"> {
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<TrackedSubsessionLink, "parentSessionId">): Record<string, unknown> {
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<string, unknown> {
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<SessionHeaderSummary | undefined> {
let file: Awaited<ReturnType<typeof open>> | 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<string | undefined> {
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<void> {
const content = await readFile(sessionFile, "utf8");
const newlineIndex = content.indexOf("\n");
@@ -1423,6 +1721,11 @@ async function clearParentSession(sessionFile: string): Promise<void> {
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";