snowspeeder/pi-web
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #31 from jmfederico/fix/persist-subsession-links

Browse Source 
fix: persist tracked subsession links
This commit is contained in:
Federico Jaramillo Martinez
2026-06-25 11:09:46 +02:00
committed by GitHub
parent 729635d246 446f588324
commit 3c09915638
5 changed files with 1117 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/persist-subsession-links.md
+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.
src/server/sessions/piSessionService.test.ts
+679 -3
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 { AuthStorage, ModelRegistry } from "@earendil-works/pi-coding-agent";
import { describe, expect, it, vi } from "vitest"; import { describe, expect, it, vi } from "vitest";
import type { GlobalSessionEvent, SessionUiEvent } from "../../shared/apiTypes.js"; import type { GlobalSessionEvent, SessionUiEvent } from "../../shared/apiTypes.js";
@@ -32,11 +35,12 @@ interface TestSession extends PiAgentSession {
getFollowUpMessages: () => readonly string[]; getFollowUpMessages: () => readonly string[];
} }
function fakeSessionManager(cwd = "/workspace"): PiSessionManager { function fakeSessionManager(cwd = "/workspace", patch: Partial<PiSessionManager> = {}): PiSessionManager {
return { return {
getCwd: () => cwd, getCwd: () => cwd,
getBranch: () => [], getBranch: () => [],
getLeafId: () => "leaf-1", 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", () => { describe("PiSessionService", () => {
it("starts sessions through an injected runtime creator", async () => { it("starts sessions through an injected runtime creator", async () => {
const hub = new CapturingSessionEventHub(); const hub = new CapturingSessionEventHub();
@@ -887,6 +901,668 @@ describe("PiSessionService", () => {
await service.dispose(); 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("does not hydrate persisted links when the exact child file is unavailable", 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(),
heartbeatIntervalMs: 60_000,
});
await service.start("/workspace");
await expect(service.listSubsessions("parent-1")).resolves.toEqual([]);
await service.dispose();
});
it("does not hydrate parent links without a child file", 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(),
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", {
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 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("notifies the validated parent file instead of an active prefix-matched parent id", async () => {
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-prefix-parent-"));
const parentFile = join(tempDir, "parent.jsonl");
const forkParentFile = join(tempDir, "parent-fork.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(forkParentFile, `${JSON.stringify({ type: "session", version: 3, id: "parent-1-fork", 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", {
getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace-feature" } }],
});
const forkManager = fakeSessionManager("/workspace");
const fork = fakeRuntime("parent-1-fork", { sessionFile: forkParentFile, sessionManager: forkManager });
const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager });
const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager });
const runtimes = [fork.runtime, child.runtime, parent.runtime];
let index = 0;
const open = vi.fn((path: string) => {
if (path === parentFile) return parentManager;
if (path === forkParentFile) return forkManager;
return childManager;
});
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: () => {
const runtime = runtimes[index] ?? parent.runtime;
index += 1;
return Promise.resolve(runtime);
},
sessionManager: {
create: () => forkManager,
list: (cwd: string) => Promise.resolve(cwd === "/workspace"
? [{ ...sessionRecord("parent-1-fork", "/workspace"), path: forkParentFile }]
: [{ ...sessionRecord("child-1", "/workspace-feature"), path: childFile, parentSessionPath: parentFile }]),
listAll: () => Promise.resolve([]),
open,
},
archiveStore: emptyArchiveStore(),
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("parent-1-fork", "/workspace"));
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(fork.calls.sendCustomMessage).toHaveLength(0);
expect(parent.calls.sendCustomMessage).toHaveLength(1);
expect(open).toHaveBeenCalledWith(parentFile);
await service.dispose();
} finally {
await rm(tempDir, { recursive: true, force: true });
}
});
it("does not relink a copied child with the original session id unless the parent link names the current child file", async () => {
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-copied-child-"));
const parentFile = join(tempDir, "parent.jsonl");
const originalChildFile = join(tempDir, "original-child.jsonl");
const copiedChildFile = join(tempDir, "copied-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(originalChildFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
await writeFile(copiedChildFile, `${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", {
getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: originalChildFile, cwd: "/workspace-feature" } }],
});
const child = fakeRuntime("child-1", { sessionFile: copiedChildFile, 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: copiedChildFile, 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(0);
await service.dispose();
} finally {
await rm(tempDir, { recursive: true, force: true });
}
});
it("uses the verified child file instead of an active copied child with the same id", async () => {
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-active-copy-child-"));
const parentFile = join(tempDir, "parent.jsonl");
const originalChildFile = join(tempDir, "original-child.jsonl");
const copiedChildFile = join(tempDir, "copied-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(originalChildFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace-feature", parentSession: parentFile })}\n`, "utf8");
await writeFile(copiedChildFile, `${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 copiedManager = fakeSessionManager("/workspace-feature", {
getBranch: () => [{ type: "message", message: { role: "assistant", content: "copied child result" } }],
});
const originalManager = fakeSessionManager("/workspace-feature", {
getBranch: () => [{ type: "message", message: { role: "assistant", content: "original child result" } }],
});
const parentManager = fakeSessionManager("/workspace", {
getEntries: () => [{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: originalChildFile, cwd: "/workspace-feature" } }],
});
const copiedChild = fakeRuntime("child-1", { sessionFile: copiedChildFile, sessionManager: copiedManager, isStreaming: true });
const originalChild = fakeRuntime("child-1", { sessionFile: originalChildFile, sessionManager: originalManager });
const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager });
const createAgentRuntime: RuntimeCreator = (_createRuntime, options) => {
if (options.sessionManager === copiedManager) return Promise.resolve(copiedChild.runtime);
if (options.sessionManager === originalManager) return Promise.resolve(originalChild.runtime);
if (options.sessionManager === parentManager) return Promise.resolve(parent.runtime);
throw new Error("unexpected session manager");
};
const open = vi.fn((path: string) => {
if (path === copiedChildFile) return copiedManager;
if (path === originalChildFile) return originalManager;
if (path === parentFile) return parentManager;
throw new Error(`unexpected open path ${path}`);
});
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime,
sessionManager: {
create: () => parentManager,
list: (cwd: string) => Promise.resolve(cwd === "/workspace-feature" ? [{ ...sessionRecord("child-1", "/workspace-feature"), path: copiedChildFile, parentSessionPath: parentFile }] : []),
listAll: () => Promise.resolve([]),
open,
},
archiveStore: emptyArchiveStore(),
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("child-1", "/workspace-feature"));
await service.start("/workspace");
await expect(service.listSubsessions("parent-1", parentFile)).resolves.toEqual([
{ sessionId: "child-1", cwd: "/workspace-feature", status: "idle" },
]);
copiedChild.session.isStreaming = true;
copiedChild.emit({ type: "agent_start" });
copiedChild.session.isStreaming = false;
copiedChild.emit({ type: "agent_end" });
await new Promise((resolve) => setTimeout(resolve, 20));
expect(parent.calls.sendCustomMessage).toHaveLength(0);
await expect(service.checkSubsession("parent-1", "child-1", parentFile)).resolves.toMatchObject({
sessionId: "child-1",
cwd: "/workspace-feature",
status: "idle",
finalText: "original child result",
messageCount: 1,
});
const read = await service.readSubsession("parent-1", "child-1", { roles: ["assistant"] }, parentFile);
expect(read.entries[0]?.parts[0]).toMatchObject({ kind: "text", text: "original child result" });
expect(open).toHaveBeenCalledWith(originalChildFile);
await service.dispose();
} finally {
await rm(tempDir, { recursive: true, force: true });
}
});
it("uses the verified parent file instead of an active copied parent with the same id", async () => {
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-active-copy-parent-"));
const parentFile = join(tempDir, "parent.jsonl");
const copiedParentFile = join(tempDir, "copied-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(copiedParentFile, `${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", {
getEntries: () => [{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } }],
getBranch: () => [{ type: "message", message: { role: "assistant", content: "child result" } }],
});
const parentManager = 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 copiedParentManager = fakeSessionManager("/workspace", { getEntries: () => [] });
const child = fakeRuntime("child-1", { sessionFile: childFile, sessionManager: childManager });
const parent = fakeRuntime("parent-1", { sessionFile: parentFile, sessionManager: parentManager });
const copiedParent = fakeRuntime("parent-1", { sessionFile: copiedParentFile, sessionManager: copiedParentManager });
const createAgentRuntime: RuntimeCreator = (_createRuntime, options) => {
if (options.sessionManager === childManager) return Promise.resolve(child.runtime);
if (options.sessionManager === parentManager) return Promise.resolve(parent.runtime);
if (options.sessionManager === copiedParentManager) return Promise.resolve(copiedParent.runtime);
throw new Error("unexpected session manager");
};
const open = vi.fn((path: string) => {
if (path === childFile) return childManager;
if (path === parentFile) return parentManager;
if (path === copiedParentFile) return copiedParentManager;
throw new Error(`unexpected open path ${path}`);
});
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime,
sessionManager: {
create: () => copiedParentManager,
list: (cwd: string) => Promise.resolve(cwd === "/workspace"
? [{ ...sessionRecord("parent-1", "/workspace"), path: copiedParentFile }]
: [{ ...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"));
await service.status(sessionRef("parent-1", "/workspace"));
await expect(service.listSubsessions("parent-1", copiedParentFile)).resolves.toEqual([]);
await expect(service.checkSubsession("parent-1", "child-1", copiedParentFile)).rejects.toThrow("not one of your subsessions");
await expect(service.readSubsession("parent-1", "child-1", {}, copiedParentFile)).rejects.toThrow("not one of your subsessions");
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(copiedParent.calls.sendCustomMessage).toHaveLength(0);
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 current child file header no longer records the parent", async () => {
const tempDir = await mkdtemp(join(tmpdir(), "pi-web-subsession-stale-child-header-"));
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 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", {
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 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(),
get: (sessionId) => Promise.resolve(sessionId === "child-1" ? { sessionId: "child-1", cwd: "/workspace-feature", archivedAt: "2026-01-01T00:00:00.000Z", parentSessionPath: parentFile } : undefined),
},
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(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 () => { it("notifies the parent once when the tracked child stops working", async () => {
const { parent, child, service } = subsessionService({ allowed: true, cwd: "/workspace-feature" }); const { parent, child, service } = subsessionService({ allowed: true, cwd: "/workspace-feature" });
await service.start("/workspace"); await service.start("/workspace");
@@ -947,7 +1623,7 @@ describe("PiSessionService", () => {
await service.dispose(); await service.dispose();
}); });
it("reports an archived child's status in the subsession list", async () => { it("reports a missing tracked child file as unknown in the subsession list", async () => {
const { service } = subsessionService({ allowed: true, cwd: "/workspace-feature" }); const { service } = subsessionService({ allowed: true, cwd: "/workspace-feature" });
await service.start("/workspace"); await service.start("/workspace");
await service.spawnSubsession({ spawningCwd: "/workspace", parentSessionId: "parent-1", parentSessionFile: "/tmp/parent-1.jsonl", prompt: "go", cwd: "/workspace-feature" }); await service.spawnSubsession({ spawningCwd: "/workspace", parentSessionId: "parent-1", parentSessionFile: "/tmp/parent-1.jsonl", prompt: "go", cwd: "/workspace-feature" });
@@ -955,7 +1631,7 @@ describe("PiSessionService", () => {
await service.archive("child-1"); await service.archive("child-1");
await expect(service.listSubsessions("parent-1")).resolves.toEqual([ await expect(service.listSubsessions("parent-1")).resolves.toEqual([
{ sessionId: "child-1", cwd: "/workspace-feature", status: "archived" }, { sessionId: "child-1", cwd: "/workspace-feature", status: "unknown" },
]); ]);
await service.dispose(); await service.dispose();
}); });
src/server/sessions/piSessionService.ts
+417 -30
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 type { Api, ImageContent, Model } from "@earendil-works/pi-ai";
import { import {
AuthStorage, AuthStorage,
@@ -81,6 +81,26 @@ interface QueuedPrompt {
echoUserMessage?: boolean; 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 { function requirePromptText(value: unknown): string {
if (typeof value !== "string") throw new Error("Prompt text is required"); if (typeof value !== "string") throw new Error("Prompt text is required");
return value; return value;
@@ -123,8 +143,10 @@ type ModelRegistryInstance = ReturnType<typeof ModelRegistry.create>;
export interface PiSessionManager { export interface PiSessionManager {
getCwd(): string; getCwd(): string;
getBranch(): unknown[]; getBranch(): unknown[];
getEntries?(): readonly unknown[];
getLeafId(): string | null; getLeafId(): string | null;
getHeader?(): { parentSession?: string } | null | undefined; getHeader?(): { parentSession?: string } | null | undefined;
appendCustomEntry?(customType: string, data?: unknown): string;
} }
export interface PiSessionManagerGateway { export interface PiSessionManagerGateway {
@@ -290,6 +312,10 @@ export class PiSessionService {
private readonly subsessionParents = new Map<string, string>(); private readonly subsessionParents = new Map<string, string>();
/** Parent session id -> the set of tracked subsession ids it spawned. */ /** Parent session id -> the set of tracked subsession ids it spawned. */
private readonly subsessionChildren = new Map<string, Set<string>>(); 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 id/file identities whose persisted links have already been loaded. */
private readonly subsessionHydratedParents = new Set<string>();
/** /**
* Tracked subsession id -> whether a completion notification is armed. * Tracked subsession id -> whether a completion notification is armed.
* Armed when the child starts working; firing on completion disarms it so a * Armed when the child starts working; firing on completion disarms it so a
@@ -322,9 +348,9 @@ export class PiSessionService {
this.spawnTargets === undefined ? undefined : (input) => this.spawnSession(input), this.spawnTargets === undefined ? undefined : (input) => this.spawnSession(input),
!subsessionsActive ? undefined : { !subsessionsActive ? undefined : {
spawn: (input) => this.spawnSubsession(input), spawn: (input) => this.spawnSubsession(input),
list: (parentSessionId) => this.listSubsessions(parentSessionId), list: (parentSessionId, parentSessionFile) => this.listSubsessions(parentSessionId, parentSessionFile),
check: (parentSessionId, sessionId) => this.checkSubsession(parentSessionId, sessionId), check: (parentSessionId, sessionId, parentSessionFile) => this.checkSubsession(parentSessionId, sessionId, parentSessionFile),
read: (parentSessionId, sessionId, query) => this.readSubsession(parentSessionId, sessionId, query), read: (parentSessionId, sessionId, query, parentSessionFile) => this.readSubsession(parentSessionId, sessionId, query, parentSessionFile),
}, },
); );
this.createAgentRuntime = deps.createAgentRuntime ?? defaultCreateAgentRuntime; this.createAgentRuntime = deps.createAgentRuntime ?? defaultCreateAgentRuntime;
@@ -362,6 +388,8 @@ export class PiSessionService {
this.authLossWarnings.clear(); this.authLossWarnings.clear();
this.subsessionParents.clear(); this.subsessionParents.clear();
this.subsessionChildren.clear(); this.subsessionChildren.clear();
this.subsessionLinks.clear();
this.subsessionHydratedParents.clear();
this.subsessionNotifyArmed.clear(); this.subsessionNotifyArmed.clear();
await Promise.all(activeSessions.map(async (active) => { await Promise.all(activeSessions.map(async (active) => {
active.unsubscribe(); active.unsubscribe();
@@ -439,7 +467,17 @@ export class PiSessionService {
const decision = await this.spawnTargets.resolveSpawnTarget(input.spawningCwd, input.cwd); const decision = await this.spawnTargets.resolveSpawnTarget(input.spawningCwd, input.cwd);
if (!decision.allowed) throw spawnTargetError(decision); if (!decision.allowed) throw spawnTargetError(decision);
const created = await this.start(decision.cwd, input.parentSessionFile); const created = await this.start(decision.cwd, input.parentSessionFile);
this.registerSubsession(input.parentSessionId, created.id); const parentSessionFile = nonEmptyString(input.parentSessionFile);
const link: TrackedSubsessionLink = {
parentSessionId: input.parentSessionId,
childSessionId: created.id,
...(created.path === "" ? {} : { childSessionFile: created.path }),
...(parentSessionFile === undefined ? {} : { parentSessionFile }),
cwd: decision.cwd,
};
this.registerVerifiedSubsession(link);
this.persistSubsessionLink(link);
this.persistSubsessionChildMarker(input.parentSessionId, created.id);
await this.prompt(created.id, input.prompt); await this.prompt(created.id, input.prompt);
this.logger.info( this.logger.info(
{ parentSessionId: input.parentSessionId, sessionId: created.id, cwd: decision.cwd, promptLength: input.prompt.length }, { parentSessionId: input.parentSessionId, sessionId: created.id, cwd: decision.cwd, promptLength: input.prompt.length },
@@ -449,65 +487,272 @@ export class PiSessionService {
} }
/** Summaries of the tracked subsessions spawned by `parentSessionId`. */ /** Summaries of the tracked subsessions spawned by `parentSessionId`. */
async listSubsessions(parentSessionId: string): Promise<SubsessionSummary[]> { async listSubsessions(parentSessionId: string, parentSessionFile?: string): Promise<SubsessionSummary[]> {
const parentFile = nonEmptyString(parentSessionFile);
await this.hydrateSubsessionsForParent(parentSessionId, parentFile);
const childIds = this.subsessionChildren.get(parentSessionId); const childIds = this.subsessionChildren.get(parentSessionId);
if (childIds === undefined) return []; if (childIds === undefined) return [];
return Promise.all([...childIds].map(async (childId) => ({ sessionId: childId, ...(await this.subsessionSummaryFields(childId)) }))); const authorizedChildIds = [...childIds].filter((childId) => this.subsessionLinkBelongsToParent(parentSessionId, parentFile, childId));
return Promise.all(authorizedChildIds.map(async (childId) => ({ sessionId: childId, ...(await this.subsessionSummaryFields(childId)) })));
} }
/** Status and final result of a subsession, scoped to the caller's children. */ /** Status and final result of a subsession, scoped to the caller's children. */
async checkSubsession(parentSessionId: string, sessionId: string): Promise<SubsessionCheckResult> { async checkSubsession(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise<SubsessionCheckResult> {
const session = await this.openSubsession(parentSessionId, sessionId); const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile);
const messages = historyMessages(session); const messages = historyMessages(session);
return { return {
sessionId, sessionId,
cwd: session.sessionManager.getCwd(), cwd: session.sessionManager.getCwd(),
status: await this.subsessionStatus(session), status: this.subsessionStatus(session),
finalText: finalAssistantText(messages), finalText: finalAssistantText(messages),
messageCount: messages.length, messageCount: messages.length,
}; };
} }
/** Filtered, paginated transcript of a subsession, scoped to the caller's children. */ /** Filtered, paginated transcript of a subsession, scoped to the caller's children. */
async readSubsession(parentSessionId: string, sessionId: string, query: SubsessionReadQuery): Promise<SubsessionReadResult> { async readSubsession(parentSessionId: string, sessionId: string, query: SubsessionReadQuery, parentSessionFile?: string): Promise<SubsessionReadResult> {
const session = await this.openSubsession(parentSessionId, sessionId); const session = await this.openSubsession(parentSessionId, sessionId, parentSessionFile);
const view = buildTranscriptView(historyMessages(session), query); const view = buildTranscriptView(historyMessages(session), query);
return { return {
sessionId, sessionId,
cwd: session.sessionManager.getCwd(), cwd: session.sessionManager.getCwd(),
status: await this.subsessionStatus(session), status: this.subsessionStatus(session),
...view, ...view,
}; };
} }
/** Open a session after verifying it is one of the caller's tracked children. */ /** Open a session after verifying it is one of the caller's tracked children. */
private async openSubsession(parentSessionId: string, sessionId: string): Promise<PiAgentSession> { private async openSubsession(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise<PiAgentSession> {
if (this.subsessionParents.get(sessionId) !== parentSessionId) { const parentFile = nonEmptyString(parentSessionFile);
await this.hydrateSubsessionsForParent(parentSessionId, parentFile);
if (this.subsessionParents.get(sessionId) !== parentSessionId || !this.subsessionLinkBelongsToParent(parentSessionId, parentFile, sessionId)) {
throw new Error(`Session ${sessionId} is not one of your subsessions`); 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 subsessionLinkBelongsToParent(parentSessionId: string, parentSessionFile: string | undefined, childSessionId: string): boolean {
const link = this.subsessionLinks.get(childSessionId);
if (link?.parentSessionId !== parentSessionId) return false;
return parentSessionFile === undefined || trackedLinkParentFileMatches(link, parentSessionFile);
}
private activeChildForSubsessionLink(link: TrackedSubsessionLink): ActiveSession<PiSessionRuntime> | undefined {
const active = this.active.get(link.childSessionId);
if (active === undefined) return undefined;
return activeSessionFileMatches(active, link.childSessionFile) ? active : undefined;
}
private activeParentForSubsessionLink(link: TrackedSubsessionLink): ActiveSession<PiSessionRuntime> | undefined {
const active = this.active.get(link.parentSessionId);
if (active === undefined) return undefined;
return activeSessionFileMatches(active, link.parentSessionFile) ? active : undefined;
}
private subsessionLinkForActiveChild(session: PiAgentSession): TrackedSubsessionLink | undefined {
const childId = session.sessionId;
const parentId = this.subsessionParents.get(childId);
const link = this.subsessionLinks.get(childId);
if (parentId === undefined || link?.parentSessionId !== parentId) return undefined;
return sessionFileMatches(session, link.childSessionFile) ? link : undefined;
}
private registerVerifiedSubsession(link: TrackedSubsessionLink): void {
const { childSessionId, parentSessionId } = link;
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); this.subsessionParents.set(childSessionId, parentSessionId);
const children = this.subsessionChildren.get(parentSessionId) ?? new Set<string>(); const children = this.subsessionChildren.get(parentSessionId) ?? new Set<string>();
children.add(childSessionId); children.add(childSessionId);
this.subsessionChildren.set(parentSessionId, children); this.subsessionChildren.set(parentSessionId, children);
this.subsessionNotifyArmed.set(childSessionId, false);
this.subsessionLinks.set(childSessionId, link);
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(link: TrackedSubsessionLink): void {
const parent = this.activeParentForSubsessionLink(link)?.runtime.session;
if (parent === undefined) return;
if (parent.sessionManager.appendCustomEntry === undefined) return;
try {
parent.sessionManager.appendCustomEntry(SUBSESSION_LINK_CUSTOM_TYPE, persistedParentSubsessionLinkData(link));
} catch (error: unknown) {
this.logger.info(
{ parentSessionId: link.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, parentSessionFile?: string): Promise<void> {
const hydrationKey = subsessionHydratedParentKey(parentSessionId, parentSessionFile);
if (this.subsessionHydratedParents.has(hydrationKey)) return;
const activeParent = this.active.get(parentSessionId);
if (activeParent !== undefined && (parentSessionFile === undefined || activeSessionFileMatches(activeParent, parentSessionFile))) {
const activeParentFile = nonEmptyString(activeParent.runtime.session.sessionFile);
await this.registerPersistedSubsessionLinks(parentSessionId, activeParent.runtime.session.sessionManager, activeParentFile);
this.subsessionHydratedParents.add(hydrationKey);
return;
}
if (parentSessionFile === undefined) return;
if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) {
this.subsessionHydratedParents.add(hydrationKey);
return;
}
let parentManager: PiSessionManager;
try {
parentManager = this.sessionManager.open(parentSessionFile);
} catch {
this.subsessionHydratedParents.add(hydrationKey);
return;
}
await this.registerPersistedSubsessionLinks(parentSessionId, parentManager, parentSessionFile);
this.subsessionHydratedParents.add(hydrationKey);
}
private async registerPersistedSubsessionLinks(parentSessionId: string, parentManager: PiSessionManager, parentSessionFile: string | undefined): Promise<void> {
// Parent custom links are the authoritative recovery record: verify the
// exact live child file/header before tracking.
const entries = parentManager.getEntries?.() ?? parentManager.getBranch();
for (const entry of entries) {
const link = parsePersistedParentSubsessionLink(entry);
if (link === undefined) continue;
const verified = await this.verifiedSubsessionLinkFromParentLink(parentSessionId, parentSessionFile, link);
if (verified === undefined) continue;
this.registerVerifiedSubsession(verified);
}
}
private async verifiedSubsessionLinkFromParentLink(parentSessionId: string, parentSessionFile: string | undefined, link: PersistedParentSubsessionLink): Promise<TrackedSubsessionLink | undefined> {
if (parentSessionFile === undefined) return undefined;
if (link.spawnedBySessionId !== parentSessionId) return undefined;
if (!(await this.parentLinkHasValidChildTarget(parentSessionFile, link))) return undefined;
return trackedSubsessionLinkFromParentLink(parentSessionId, link, parentSessionFile);
}
private async parentLinkHasValidChildTarget(parentSessionFile: string, link: PersistedParentSubsessionLink): Promise<boolean> {
return link.spawnedSessionFile !== undefined
&& await sessionFileHeaderMatches(link.spawnedSessionFile, { sessionId: link.spawnedSessionId, parentSessionFile });
}
private async recoverSubsessionTrackingForOpenedSession(session: PiAgentSession): Promise<void> {
const link = await this.verifiedSubsessionLinkFromOpenedChild(session);
if (link === undefined) return;
this.registerVerifiedSubsession(link);
}
private async verifiedSubsessionLinkFromOpenedChild(session: PiAgentSession): Promise<TrackedSubsessionLink | undefined> {
// Child markers are only hints; the current child header and reciprocal
// parent custom link must agree on the exact ids and files before relinking.
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 undefined;
const childSessionFile = nonEmptyString(session.sessionFile);
if (childSessionFile === undefined) return undefined;
const childHeader = await readSessionHeaderSummary(childSessionFile);
if (childHeader?.id !== session.sessionId) return undefined;
const parentSessionFile = nonEmptyString(childHeader.parentSession);
if (parentSessionFile === undefined) return undefined;
const parentHeader = await readSessionHeaderSummary(parentSessionFile);
if (parentHeader?.id !== marker.spawnedBySessionId) return undefined;
const parentLink = this.findReciprocalParentSubsessionLink(parentSessionFile, marker.spawnedBySessionId, session.sessionId, childSessionFile);
if (parentLink === undefined) return undefined;
return {
parentSessionId: marker.spawnedBySessionId,
childSessionId: session.sessionId,
childSessionFile,
parentSessionFile,
cwd: parentLink.cwd ?? session.sessionManager.getCwd(),
};
}
private findReciprocalParentSubsessionLink(parentSessionFile: string, parentSessionId: string, childSessionId: string, childSessionFile: string): PersistedParentSubsessionLink | undefined {
let parentManager: PiSessionManager;
try {
parentManager = this.sessionManager.open(parentSessionFile);
} catch {
return undefined;
}
const entries = parentManager.getEntries?.() ?? parentManager.getBranch();
for (const entry of entries) {
const link = parsePersistedParentSubsessionLink(entry);
if (link === undefined) continue;
if (link.spawnedBySessionId !== parentSessionId || link.spawnedSessionId !== childSessionId) continue;
if (link.spawnedSessionFile === undefined || !sessionPathsEqual(link.spawnedSessionFile, childSessionFile)) continue;
return link;
}
return undefined;
}
private async getOrOpenTrackedSubsession(sessionId: string): Promise<PiAgentSession> {
const link = this.subsessionLinks.get(sessionId);
if (link === undefined) throw new Error("Session not found");
const active = this.activeChildForSubsessionLink(link);
if (active !== undefined) return active.runtime.session;
if (link.childSessionFile !== undefined) {
if (!(await sessionFileHeaderMatches(link.childSessionFile, { sessionId, parentSessionFile: link.parentSessionFile }))) throw new Error("Session not found");
const sessionManager = this.sessionManager.open(link.childSessionFile);
return (await this.create(sessionManager, link.cwd ?? sessionManager.getCwd())).runtime.session;
}
throw new Error("Session not found");
} }
private async subsessionSummaryFields(childSessionId: string): Promise<{ cwd: string; status: SubsessionStatus }> { private async subsessionSummaryFields(childSessionId: string): Promise<{ cwd: string; status: SubsessionStatus }> {
const active = this.active.get(childSessionId); const link = this.subsessionLinks.get(childSessionId);
const active = link === undefined ? undefined : this.activeChildForSubsessionLink(link);
if (active !== undefined) { if (active !== undefined) {
return { cwd: active.runtime.cwd, status: await this.subsessionStatus(active.runtime.session) }; return { cwd: active.runtime.cwd, status: this.subsessionStatus(active.runtime.session) };
} }
const archived = await this.archiveStore.get(childSessionId); if (link?.childSessionFile !== undefined && (await sessionFileHeaderMatches(link.childSessionFile, { sessionId: childSessionId, parentSessionFile: link.parentSessionFile }))) {
if (archived !== undefined) return { cwd: archived.cwd, status: "archived" }; return { cwd: link.cwd ?? "", status: "idle" };
}
if (link?.cwd !== undefined) return { cwd: link.cwd, status: "unknown" };
return { cwd: "", status: "unknown" }; return { cwd: "", status: "unknown" };
} }
private async subsessionStatus(session: PiAgentSession): Promise<SubsessionStatus> { private subsessionStatus(session: PiAgentSession): SubsessionStatus {
if (await this.archiveStore.isArchived(session.sessionId)) return "archived";
if (this.hasActiveWork(session)) return "working"; if (this.hasActiveWork(session)) return "working";
if (this.activities.get(session.sessionId)?.phase === "error") return "error"; if (this.activities.get(session.sessionId)?.phase === "error") return "error";
return "idle"; return "idle";
@@ -520,9 +765,9 @@ export class PiSessionService {
* parent is busy and delivers immediately when it is idle). * parent is busy and delivers immediately when it is idle).
*/ */
private updateSubsessionTracking(session: PiAgentSession): void { private updateSubsessionTracking(session: PiAgentSession): void {
const childId = session.sessionId; const link = this.subsessionLinkForActiveChild(session);
const parentId = this.subsessionParents.get(childId); if (link === undefined) return;
if (parentId === undefined) return; const childId = link.childSessionId;
if (this.hasActiveWork(session)) { if (this.hasActiveWork(session)) {
this.subsessionNotifyArmed.set(childId, true); this.subsessionNotifyArmed.set(childId, true);
return; return;
@@ -533,7 +778,23 @@ export class PiSessionService {
const finalText = finalAssistantText(historyMessages(session)); const finalText = finalAssistantText(historyMessages(session));
const preview = finalText === "" ? "(no output)" : truncateForNotification(finalText); const preview = finalText === "" ? "(no output)" : truncateForNotification(finalText);
const text = `Subsession ${childId} stopped working (status: ${status}). Latest output:\n\n${preview}\n\nUse check_subsession with sessionId "${childId}" for its status and latest output, or read_subsession to look through its full transcript.`; const text = `Subsession ${childId} stopped working (status: ${status}). Latest output:\n\n${preview}\n\nUse check_subsession with sessionId "${childId}" for its status and latest output, or read_subsession to look through its full transcript.`;
void this.notifyParentOfSubsession(parentId, childId, text); void this.notifyParentOfSubsession(link.parentSessionId, childId, text);
}
private async getOrOpenParentForSubsession(parentSessionId: string, childSessionId: string): Promise<PiAgentSession> {
const link = this.subsessionLinks.get(childSessionId);
if (link?.parentSessionId !== parentSessionId) throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
const active = this.activeParentForSubsessionLink(link);
if (active !== undefined) return active.runtime.session;
const parentSessionFile = link.parentSessionFile;
if (parentSessionFile === undefined) throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
if ((await readSessionHeaderSummary(parentSessionFile))?.id !== parentSessionId) {
throw new Error(`Parent session ${parentSessionId} is not available for subsession notification`);
}
const sessionManager = this.sessionManager.open(parentSessionFile);
return (await this.create(sessionManager, sessionManager.getCwd())).runtime.session;
} }
/** /**
@@ -545,7 +806,7 @@ export class PiSessionService {
*/ */
private async notifyParentOfSubsession(parentId: string, childId: string, text: string): Promise<void> { private async notifyParentOfSubsession(parentId: string, childId: string, text: string): Promise<void> {
try { try {
const session = await this.getOrOpen(parentId); const session = await this.getOrOpenParentForSubsession(parentId, childId);
await session.sendCustomMessage( await session.sendCustomMessage(
{ customType: SUBSESSION_NOTIFICATION_CUSTOM_TYPE, content: text, display: true, details: { sessionId: childId } }, { customType: SUBSESSION_NOTIFICATION_CUSTOM_TYPE, content: text, display: true, details: { sessionId: childId } },
{ triggerTurn: true, deliverAs: "followUp" }, { triggerTurn: true, deliverAs: "followUp" },
@@ -809,6 +1070,8 @@ export class PiSessionService {
const sessionFile = session.sessionFile; const sessionFile = session.sessionFile;
if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted");
await clearParentSession(sessionFile); await clearParentSession(sessionFile);
clearParentSessionHeader(session.sessionManager);
this.unregisterSubsession(session.sessionId);
} }
async abort(ref: PiSessionLookup): Promise<void> { async abort(ref: PiSessionLookup): Promise<void> {
@@ -922,7 +1185,7 @@ export class PiSessionService {
// Disarm subsession notification before teardown so the abort below cannot // Disarm subsession notification before teardown so the abort below cannot
// emit a "stopped working" event that notifies the parent (e.g. on archive). // emit a "stopped working" event that notifies the parent (e.g. on archive).
// The parent/children link is kept so the parent can still see the child. // The parent/children link is kept so the parent can still see the child.
this.subsessionNotifyArmed.delete(sessionId); if (this.subsessionLinkForActiveChild(active.runtime.session) !== undefined) this.subsessionNotifyArmed.delete(sessionId);
clearSessionQueue(active.runtime.session); clearSessionQueue(active.runtime.session);
active.unsubscribe(); active.unsubscribe();
try { try {
@@ -979,8 +1242,10 @@ export class PiSessionService {
runtime.setRebindSession(async (session) => { runtime.setRebindSession(async (session) => {
await this.bindSessionExtensions(session); await this.bindSessionExtensions(session);
this.bindRuntime(active); this.bindRuntime(active);
await this.recoverSubsessionTrackingForOpenedSession(session);
}); });
this.active.set(runtime.session.sessionId, active); this.active.set(runtime.session.sessionId, active);
await this.recoverSubsessionTrackingForOpenedSession(runtime.session);
this.publishStatus(runtime.session); this.publishStatus(runtime.session);
return active; return active;
} }
@@ -1411,6 +1676,117 @@ function isDefined<T>(value: T | undefined): value is T {
return value !== undefined; return value !== undefined;
} }
function trackedSubsessionLinkFromParentLink(parentSessionId: string, link: PersistedParentSubsessionLink, parentSessionFile: string): TrackedSubsessionLink {
return {
parentSessionId,
childSessionId: link.spawnedSessionId,
...(link.spawnedSessionFile === undefined ? {} : { childSessionFile: link.spawnedSessionFile }),
parentSessionFile,
...(link.cwd === undefined ? {} : { cwd: link.cwd }),
};
}
function persistedParentSubsessionLinkData(link: TrackedSubsessionLink): Record<string, unknown> {
return {
version: 1,
spawnedBySessionId: link.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 subsessionHydratedParentKey(parentSessionId: string, parentSessionFile: string | undefined): string {
return `${parentSessionId}\0${parentSessionFile ?? ""}`;
}
function sessionPathsEqual(a: string, b: string): boolean {
return cwdPathsEqual(a, b);
}
function sessionFileMatches(session: PiAgentSession, expectedSessionFile: string | undefined): boolean {
const sessionFile = nonEmptyString(session.sessionFile);
return sessionFile !== undefined && expectedSessionFile !== undefined && sessionPathsEqual(sessionFile, expectedSessionFile);
}
function activeSessionFileMatches(active: ActiveSession<PiSessionRuntime>, expectedSessionFile: string | undefined): boolean {
return sessionFileMatches(active.runtime.session, expectedSessionFile);
}
function trackedLinkParentFileMatches(link: TrackedSubsessionLink, parentSessionFile: string): boolean {
return link.parentSessionFile !== undefined && sessionPathsEqual(link.parentSessionFile, parentSessionFile);
}
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 sessionFileHeaderMatches(sessionFile: string, expected: { sessionId: string; parentSessionFile?: string | undefined }): Promise<boolean> {
const header = await readSessionHeaderSummary(sessionFile);
if (header?.id !== expected.sessionId) return false;
if (expected.parentSessionFile === undefined) return true;
return header.parentSession !== undefined && sessionPathsEqual(header.parentSession, expected.parentSessionFile);
}
async function clearParentSession(sessionFile: string): Promise<void> { async function clearParentSession(sessionFile: string): Promise<void> {
const content = await readFile(sessionFile, "utf8"); const content = await readFile(sessionFile, "utf8");
const newlineIndex = content.indexOf("\n"); const newlineIndex = content.indexOf("\n");
@@ -1423,6 +1799,11 @@ async function clearParentSession(sessionFile: string): Promise<void> {
await writeFile(sessionFile, `${JSON.stringify(header)}${rest}`, "utf8"); 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 { function clearSessionQueue(session: PiAgentSession): void {
session.clearQueue(); session.clearQueue();
} }
@@ -1475,6 +1856,12 @@ function historyMessages(session: PiAgentSession): unknown[] {
return messages; 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. */ /** customType marking a parent-facing subsession-completion notice. */
const SUBSESSION_NOTIFICATION_CUSTOM_TYPE = "subsession.completion"; const SUBSESSION_NOTIFICATION_CUSTOM_TYPE = "subsession.completion";
src/server/sessions/spawnSubsessionTool.test.ts
+6 -6
View File
@@ -56,9 +56,9 @@ describe("createSubsessionToolDefinitions", () => {
])); ]));
const { list: listTool } = tools({ list }); const { list: listTool } = tools({ list });
const result = await listTool.execute("call-2", {}, undefined, undefined, ctxFor("parent-1", undefined)); const result = await listTool.execute("call-2", {}, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl"));
expect(list).toHaveBeenCalledWith("parent-1"); expect(list).toHaveBeenCalledWith("parent-1", "/sessions/parent-1.jsonl");
expect(result.details).toEqual({ subsessions: [ expect(result.details).toEqual({ subsessions: [
{ sessionId: "child-1", cwd: "/repos/a", status: "working" }, { sessionId: "child-1", cwd: "/repos/a", status: "working" },
{ sessionId: "child-2", cwd: "/repos/a", status: "idle" }, { sessionId: "child-2", cwd: "/repos/a", status: "idle" },
@@ -76,9 +76,9 @@ describe("createSubsessionToolDefinitions", () => {
const check = vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/repos/a", status: "idle" as const, finalText: "all done", messageCount: 4 })); const check = vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/repos/a", status: "idle" as const, finalText: "all done", messageCount: 4 }));
const { check: checkTool } = tools({ check }); const { check: checkTool } = tools({ check });
const result = await checkTool.execute("call-4", { sessionId: "child-1" }, undefined, undefined, ctxFor("parent-1", undefined)); const result = await checkTool.execute("call-4", { sessionId: "child-1" }, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl"));
expect(check).toHaveBeenCalledWith("parent-1", "child-1"); expect(check).toHaveBeenCalledWith("parent-1", "child-1", "/sessions/parent-1.jsonl");
expect(result.details).toMatchObject({ sessionId: "child-1", status: "idle", finalText: "all done" }); expect(result.details).toMatchObject({ sessionId: "child-1", status: "idle", finalText: "all done" });
expect(firstText(result.content)).toContain("all done"); expect(firstText(result.content)).toContain("all done");
}); });
@@ -99,9 +99,9 @@ describe("createSubsessionToolDefinitions", () => {
})); }));
const { read: readTool } = tools({ read }); const { read: readTool } = tools({ read });
const result = await readTool.execute("call-6", { sessionId: "child-1", roles: ["assistant"], maxChars: 200 }, undefined, undefined, ctxFor("parent-1", undefined)); const result = await readTool.execute("call-6", { sessionId: "child-1", roles: ["assistant"], maxChars: 200 }, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl"));
expect(read).toHaveBeenCalledWith("parent-1", "child-1", { roles: ["assistant"], maxChars: 200 }); expect(read).toHaveBeenCalledWith("parent-1", "child-1", { roles: ["assistant"], maxChars: 200 }, "/sessions/parent-1.jsonl");
expect(result.details).toMatchObject({ sessionId: "child-1", matched: 1 }); expect(result.details).toMatchObject({ sessionId: "child-1", matched: 1 });
expect(firstText(result.content)).toContain("the answer"); expect(firstText(result.content)).toContain("the answer");
}); });
src/server/sessions/spawnSubsessionTool.ts
+10 -7
View File
@@ -3,7 +3,7 @@ import { defineTool } from "@earendil-works/pi-coding-agent";
import type { TranscriptContentKind, TranscriptEntry, TranscriptRole, TranscriptView } from "./subsessionTranscript.js"; import type { TranscriptContentKind, TranscriptEntry, TranscriptRole, TranscriptView } from "./subsessionTranscript.js";
/** Lifecycle phase of a tracked subsession as seen by its parent. */ /** Lifecycle phase of a tracked subsession as seen by its parent. */
export type SubsessionStatus = "working" | "idle" | "error" | "archived" | "unknown"; export type SubsessionStatus = "working" | "idle" | "error" | "unknown";
export interface SpawnSubsessionResult { export interface SpawnSubsessionResult {
sessionId: string; sessionId: string;
@@ -56,9 +56,9 @@ export interface SubsessionReadQuery {
export interface SubsessionToolDeps { export interface SubsessionToolDeps {
spawn(input: SpawnSubsessionInvocation): Promise<SpawnSubsessionResult>; spawn(input: SpawnSubsessionInvocation): Promise<SpawnSubsessionResult>;
list(parentSessionId: string): Promise<SubsessionSummary[]>; list(parentSessionId: string, parentSessionFile?: string): Promise<SubsessionSummary[]>;
check(parentSessionId: string, sessionId: string): Promise<SubsessionCheckResult>; check(parentSessionId: string, sessionId: string, parentSessionFile?: string): Promise<SubsessionCheckResult>;
read(parentSessionId: string, sessionId: string, query: SubsessionReadQuery): Promise<SubsessionReadResult>; read(parentSessionId: string, sessionId: string, query: SubsessionReadQuery, parentSessionFile?: string): Promise<SubsessionReadResult>;
} }
const SpawnSubsessionParams = Type.Object({ const SpawnSubsessionParams = Type.Object({
@@ -196,7 +196,8 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
parameters: ListSubsessionsParams, parameters: ListSubsessionsParams,
async execute(_toolCallId, _params, _signal, _onUpdate, ctx) { async execute(_toolCallId, _params, _signal, _onUpdate, ctx) {
const parentSessionId = ctx.sessionManager.getSessionId(); const parentSessionId = ctx.sessionManager.getSessionId();
const subsessions = await deps.list(parentSessionId); const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
const subsessions = await deps.list(parentSessionId, parentSessionFile);
const text = subsessions.length === 0 const text = subsessions.length === 0
? "You have not spawned any subsessions." ? "You have not spawned any subsessions."
: `Your subsessions:\n${subsessions.map(statusLine).join("\n")}`; : `Your subsessions:\n${subsessions.map(statusLine).join("\n")}`;
@@ -212,7 +213,8 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
parameters: CheckSubsessionParams, parameters: CheckSubsessionParams,
async execute(_toolCallId, params, _signal, _onUpdate, ctx) { async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
const parentSessionId = ctx.sessionManager.getSessionId(); const parentSessionId = ctx.sessionManager.getSessionId();
const result = await deps.check(parentSessionId, params.sessionId); const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
const result = await deps.check(parentSessionId, params.sessionId, parentSessionFile);
const body = result.finalText === "" ? "(no output yet)" : result.finalText; const body = result.finalText === "" ? "(no output yet)" : result.finalText;
return { return {
content: [{ type: "text", text: `Subsession ${result.sessionId} [${result.status}]:\n\n${body}` }], content: [{ type: "text", text: `Subsession ${result.sessionId} [${result.status}]:\n\n${body}` }],
@@ -229,8 +231,9 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
parameters: ReadSubsessionParams, parameters: ReadSubsessionParams,
async execute(_toolCallId, params, _signal, _onUpdate, ctx) { async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
const parentSessionId = ctx.sessionManager.getSessionId(); const parentSessionId = ctx.sessionManager.getSessionId();
const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
const { sessionId, ...query } = params; const { sessionId, ...query } = params;
const result = await deps.read(parentSessionId, sessionId, query); const result = await deps.read(parentSessionId, sessionId, query, parentSessionFile);
return { return {
content: [{ type: "text", text: renderTranscript(result) }], content: [{ type: "text", text: renderTranscript(result) }],
details: result, details: result,