Merge remote-tracking branch 'origin/main' into pr-36-generic-agent-config

# Conflicts:
#	docs/config.html
#	docs/config.md
#	src/cli.test.ts
#	src/cli.ts
#	src/client/src/components/settings/SettingsSessiondPanel.ts
#	src/client/src/components/settings/settingsConfigDraft.test.ts
#	src/client/src/components/settings/settingsConfigDraft.ts
#	src/server/app.test.ts
#	src/server/app.ts
#	src/server/configRoutes.test.ts
#	src/server/configRoutes.ts
#	src/server/piWebPluginService.test.ts
#	src/server/piWebPluginService.ts
#	src/server/piWebStatus.test.ts
#	src/server/piWebStatus.ts
#	src/server/piWebStatusCache.ts
#	src/server/sessions/authService.test.ts
#	src/server/sessions/piSessionService.ts
#	src/server/sessions/sessionRoutes.test.ts
This commit is contained in:
Federico Jaramillo Martinez
2026-07-13 20:30:08 +02:00
289 changed files with 29164 additions and 6419 deletions
+89 -3
View File
@@ -1,13 +1,21 @@
import { mkdir, mkdtemp, readFile, readdir, rm, symlink } from "node:fs/promises";
import { join } from "node:path";
import { basename, join } from "node:path";
import { tmpdir } from "node:os";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { DEFAULT_ATTACHMENT_FOLDER, saveAttachmentsToWorkspace } from "./attachmentService.js";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { formatDimensionNote, resizeImage, type ResizedImage } from "@earendil-works/pi-coding-agent";
import { DEFAULT_ATTACHMENT_FOLDER, attachmentsToInlineImages, saveAttachmentsToWorkspace } from "./attachmentService.js";
vi.mock("@earendil-works/pi-coding-agent", () => ({
formatDimensionNote: vi.fn(),
resizeImage: vi.fn(),
}));
let workspace: string;
let externalDirectories: string[] = [];
beforeEach(async () => {
vi.mocked(formatDimensionNote).mockReset();
vi.mocked(resizeImage).mockReset();
workspace = await mkdtemp(join(tmpdir(), "pi-web-attachments-"));
externalDirectories = [];
});
@@ -22,6 +30,63 @@ afterEach(async () => {
const pngBytes = Buffer.from([0x89, 0x50, 0x4e, 0x47]);
const pngBase64 = pngBytes.toString("base64");
function resizedImage(overrides: Partial<ResizedImage> = {}): ResizedImage {
return {
data: "resized-data",
mimeType: "image/png",
originalWidth: 2400,
originalHeight: 1200,
width: 1200,
height: 600,
wasResized: true,
...overrides,
};
}
describe("attachmentsToInlineImages", () => {
it("resizes images, drops unresizable images, and preserves dimension notes", async () => {
const firstInput = Buffer.from("first image");
const droppedInput = Buffer.from("too large");
const thirdInput = Buffer.from("third image");
const firstResized = resizedImage({ data: "first-resized", mimeType: "image/webp" });
const thirdResized = resizedImage({
data: "third-resized",
mimeType: "image/jpeg",
originalWidth: 640,
originalHeight: 480,
width: 640,
height: 480,
wasResized: false,
});
vi.mocked(resizeImage)
.mockResolvedValueOnce(firstResized)
.mockResolvedValueOnce(null)
.mockResolvedValueOnce(thirdResized);
vi.mocked(formatDimensionNote)
.mockReturnValueOnce("[Image dimensions changed.]")
.mockReturnValueOnce(undefined);
await expect(attachmentsToInlineImages([
{ kind: "image", mimeType: "image/png", data: firstInput.toString("base64"), name: "first.png" },
{ kind: "image", mimeType: "image/png", data: droppedInput.toString("base64"), name: "huge.png" },
{ kind: "image", mimeType: "image/jpeg", data: thirdInput.toString("base64"), name: "photo.jpg" },
])).resolves.toEqual([
{
image: { type: "image", data: "first-resized", mimeType: "image/webp" },
dimensionNote: "[Image dimensions changed.]",
},
{ image: { type: "image", data: "third-resized", mimeType: "image/jpeg" } },
]);
expect(resizeImage).toHaveBeenNthCalledWith(1, firstInput, "image/png");
expect(resizeImage).toHaveBeenNthCalledWith(2, droppedInput, "image/png");
expect(resizeImage).toHaveBeenNthCalledWith(3, thirdInput, "image/jpeg");
expect(formatDimensionNote).toHaveBeenNthCalledWith(1, firstResized);
expect(formatDimensionNote).toHaveBeenNthCalledWith(2, thirdResized);
});
});
describe("saveAttachmentsToWorkspace", () => {
it("writes attachments into the default folder and returns relative paths", async () => {
const fixedNow = () => new Date("2026-06-13T12:05:01.123Z");
@@ -69,6 +134,27 @@ describe("saveAttachmentsToWorkspace", () => {
expect(await readFile(join(workspace, saved[1]?.path ?? ""))).toHaveLength(0);
});
it("falls back, strips controls, and truncates unsafe attachment names", async () => {
const longStem = "a".repeat(140);
const saved = await saveAttachmentsToWorkspace(
workspace,
[
{ kind: "image", mimeType: "image/jpeg", data: pngBase64 },
{ kind: "file", mimeType: "application/octet-stream", data: "QUJD", name: "\u0000\u001f\u007f" },
{ kind: "file", mimeType: "text/plain", data: "REVG", name: "nested/bad\u0000\u007fname\n.txt" },
{ kind: "file", mimeType: "application/pdf", data: "R0hJ", name: `${longStem}.pdf` },
],
{ now: () => new Date(2026, 5, 13, 12, 5, 1, 123) },
);
expect(saved.map((attachment) => basename(attachment.path))).toEqual([
"attachment-20260613-120501-123-1-image.jpg",
"attachment-20260613-120501-123-2-file.bin",
"attachment-20260613-120501-123-3-badname.txt",
`attachment-20260613-120501-123-4-${"a".repeat(92)}.pdf`,
]);
});
it("does not overwrite an existing attachment name", async () => {
const fixedNow = () => new Date("2026-06-13T12:05:01.123Z");
const first = await saveAttachmentsToWorkspace(
@@ -33,7 +33,7 @@ describe("auth provider options", () => {
expect(isApiKeyLoginProvider("openai", new Set(["openai-codex"]))).toBe(true);
});
it("includes Anthropic in both OAuth and API key login options", () => {
it("builds login options for OAuth-only, dual-auth, and API-key providers", () => {
const options = getLoginProviderOptions(registry());
expect(options).toEqual(expect.arrayContaining([
expect.objectContaining({ id: "anthropic", authType: "oauth" }),
@@ -44,7 +44,7 @@ describe("auth provider options", () => {
expect(options).not.toEqual(expect.arrayContaining([expect.objectContaining({ id: "openai-codex", authType: "api_key" })]));
});
it("returns only stored credentials for logout", () => {
it("returns only currently stored credentials for logout", () => {
expect(getLogoutProviderOptions(registry())).toEqual([
expect.objectContaining({ id: "openai", authType: "api_key" }),
]);
+50 -1
View File
@@ -2,8 +2,10 @@ import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { AuthStorage, ModelRegistry } from "@earendil-works/pi-coding-agent";
import { afterEach, describe, expect, it } from "vitest";
import { afterEach, describe, expect, it, vi } from "vitest";
import type { OAuthFlowState } from "../../shared/apiTypes.js";
import { AuthService, type AuthChange } from "./authService.js";
import { OAuthLoginFlowService } from "./oauthLoginFlowService.js";
const tempDirs: string[] = [];
@@ -49,6 +51,39 @@ describe("AuthService", () => {
await expect(readFile(join(agentDir, "auth.json"), "utf8")).resolves.toContain("sk-test");
auth.dispose();
});
it("refreshes auth state after OAuth login completes", () => {
const authStorage = AuthStorage.inMemory();
const modelRegistry = ModelRegistry.create(authStorage);
const authFlows = new CapturingOAuthLoginFlowService();
const auth = new AuthService({ modelRegistry, authFlows });
const changes: AuthChange[] = [];
auth.subscribe((change) => { changes.push(change); });
const reload = vi.spyOn(authStorage, "reload");
const refresh = vi.spyOn(modelRegistry, "refresh");
const provider = authStorage.getOAuthProviders().find((option) => option.id === "anthropic");
if (provider === undefined) throw new Error("Expected built-in OAuth provider");
expect(auth.startOAuthLogin(provider.id)).toMatchObject({ providerId: provider.id, providerName: provider.name, status: "running" });
const startOptions = authFlows.startCalls.at(0);
if (startOptions === undefined) throw new Error("Expected OAuth flow to start");
expect(startOptions.providerId).toBe(provider.id);
expect(startOptions.providerName).toBe(provider.name);
expect(startOptions.authStorage).toBe(authStorage);
expect(changes).toEqual([]);
reload.mockClear();
refresh.mockClear();
if (startOptions.onComplete === undefined) throw new Error("Expected OAuth completion callback");
startOptions.onComplete();
expect(reload).toHaveBeenCalledOnce();
expect(refresh).toHaveBeenCalledOnce();
expect(changes).toEqual([{}]);
auth.dispose();
expect(authFlows.disposed).toBe(true);
});
});
function createAuthService(data: Parameters<typeof AuthStorage.inMemory>[0] = {}) {
@@ -65,3 +100,17 @@ async function tempAgentDir(): Promise<string> {
tempDirs.push(dir);
return dir;
}
class CapturingOAuthLoginFlowService extends OAuthLoginFlowService {
readonly startCalls: Parameters<OAuthLoginFlowService["start"]>[0][] = [];
disposed = false;
override start(options: Parameters<OAuthLoginFlowService["start"]>[0]): OAuthFlowState {
this.startCalls.push(options);
return { flowId: "flow-1", providerId: options.providerId, providerName: options.providerName, status: "running", progress: [] };
}
override dispose(): void {
this.disposed = true;
}
}
+1 -1
View File
@@ -20,7 +20,7 @@ export const BUILTIN_COMMANDS: ClientCommand[] = [
{ name: "new", description: "Start a new session", source: "builtin" },
{ name: "compact", description: "Manually compact session context", source: "builtin" },
{ name: "resume", description: "Resume a different session", source: "builtin" },
{ name: "reload", description: "Reload keybindings, extensions, skills, prompts, and themes", source: "builtin" },
{ name: "reload", description: "Reload Pi runtime resources for this session", source: "builtin" },
{ name: "quit", description: "Quit pi", source: "builtin" },
];
@@ -12,6 +12,7 @@ afterEach(() => {
describe("OAuthLoginFlowService", () => {
it("round-trips prompt responses and completes the flow", async () => {
let promptValue: string | undefined;
const onComplete = vi.fn();
const service = new OAuthLoginFlowService();
const state = service.start({
providerId: "test-provider",
@@ -22,6 +23,7 @@ describe("OAuthLoginFlowService", () => {
promptValue = await callbacks.onPrompt({ message: "Paste code", placeholder: "code" });
callbacks.onProgress?.(`Got ${promptValue}`);
}),
onComplete,
});
const prompt = state.prompt;
@@ -35,6 +37,7 @@ describe("OAuthLoginFlowService", () => {
expect(promptValue).toBe("abc123");
expect(service.get(state.flowId)).toMatchObject({ status: "complete", progress: ["Waiting for code", "Got abc123", "Login complete"] });
expect(onComplete).toHaveBeenCalledOnce();
service.dispose();
});
@@ -113,6 +116,30 @@ describe("OAuthLoginFlowService", () => {
service.dispose();
});
it("rejects pending prompts when disposed", async () => {
const promptRejected = deferred<Error>();
const service = new OAuthLoginFlowService();
const state = service.start({
providerId: "test-provider",
providerName: "Test Provider",
authStorage: fakeAuthStorage(async (_providerId, callbacks) => {
try {
await callbacks.onPrompt({ message: "Paste code" });
} catch (error) {
promptRejected.resolve(toError(error));
throw error;
}
}),
});
expect(state.prompt).toBeDefined();
service.dispose();
await expect(promptRejected.promise).resolves.toMatchObject({ message: "Login cancelled" });
expect(() => { service.get(state.flowId); }).toThrow("OAuth login flow not found");
});
it("rejects stale or duplicate responses", () => {
const service = new OAuthLoginFlowService();
const state = service.start({
@@ -0,0 +1,358 @@
import { describe, expect, it, vi } from "vitest";
import { PiSessionService } from "./piSessionService.js";
import { CapturingSessionEventHub, fakeRuntime, fakeSessionManager, runtimeCreator, sessionGateway, sessionRecord, sessionRef } from "./piSessionService.testSupport.js";
describe("PiSessionService archive and cleanup", () => {
it("archives a session subtree within the root workspace", async () => {
const archivedInputs: string[] = [];
const root = sessionRecord("root");
const directChild = { ...sessionRecord("direct-child"), path: "/sessions/direct-child.jsonl", parentSessionPath: root.path };
const archivedChild = { ...sessionRecord("archived-child"), path: "/sessions/archived-child.jsonl", parentSessionPath: root.path };
const grandchild = { ...sessionRecord("grandchild"), path: "/sessions/grandchild.jsonl", parentSessionPath: archivedChild.path };
const otherWorkspaceChild = { ...sessionRecord("other-child", "/other"), path: "/sessions/other-child.jsonl", parentSessionPath: root.path };
const fake = fakeRuntime("root", { sessionFile: root.path });
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
archiveStore: {
list: () => Promise.resolve([{ sessionId: "archived-child", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", originalPath: archivedChild.path, archivePath: "/archive/archived-child.jsonl", created: "2026-01-01T00:00:00.000Z", modified: "2026-01-01T00:01:00.000Z", messageCount: 1, firstMessage: "archived", parentSessionPath: root.path }]),
get: () => Promise.resolve(undefined),
archive: (input) => {
archivedInputs.push(input.sessionId);
return Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" });
},
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager(),
list: (cwd) => Promise.resolve(cwd === "/workspace" ? [root, directChild, archivedChild, grandchild] : [otherWorkspaceChild]),
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
await expect(service.archiveTree(sessionRef("root"))).resolves.toEqual({
archived: true,
sessionIds: ["root", "direct-child", "grandchild"],
archivedCount: 3,
skippedAlreadyArchivedCount: 1,
});
expect(archivedInputs).toEqual(["root", "direct-child", "grandchild"]);
await service.dispose();
});
it("permanently deletes archived sessions through the archive store", async () => {
const deletedSessionIds: string[] = [];
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([]),
get: (sessionId) => Promise.resolve(sessionId === "archived" || "archived".startsWith(sessionId)
? { sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/archived.jsonl" }
: undefined),
archive: () => { throw new Error("archive should not be called for records that already have archive files"); },
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
deleteArchived: (sessionId) => {
deletedSessionIds.push(sessionId);
return Promise.resolve();
},
},
sessionManager: sessionGateway([sessionRecord("active")]),
heartbeatIntervalMs: 60_000,
});
await expect(service.deleteArchived("arch")).resolves.toBeUndefined();
await expect(service.deleteArchived("active")).rejects.toThrow("Archived session not found");
expect(deletedSessionIds).toEqual(["archived"]);
await service.dispose();
});
it("bulk archives inactive sessions by cwd without opening runtimes", async () => {
const recordsByCwd = new Map([
["/one", [sessionRecord("a", "/one"), sessionRecord("b", "/one")]],
["/two", [sessionRecord("c", "/two")]],
]);
const listCalls: string[] = [];
const open = vi.fn(() => { throw new Error("bulk archive should not open inactive runtimes"); });
const archiveMany = vi.fn((inputs: readonly { sessionId: string; cwd: string }[]) => Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" }))));
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([]),
get: () => Promise.resolve(undefined),
archive: (input) => Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" }),
archiveMany,
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager(),
list: (cwd) => {
listCalls.push(cwd);
return Promise.resolve(recordsByCwd.get(cwd) ?? []);
},
open,
},
heartbeatIntervalMs: 60_000,
});
const result = await service.archiveMany([{ id: "a", cwd: "/one" }, { id: "b", cwd: "/one" }, { id: "c", cwd: "/two" }]);
expect(result).toMatchObject({ archived: true, archivedSessionIds: ["a", "b", "c"], failures: [] });
expect(listCalls).toEqual(["/one", "/two"]);
expect(open).not.toHaveBeenCalled();
expect(archiveMany).toHaveBeenCalledTimes(1);
expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["a", "b", "c"]);
await service.dispose();
});
it("bulk archive reports per-session failures without aborting other archives", async () => {
const busy = fakeRuntime("busy", { isStreaming: true });
let createCalls = 0;
const archiveMany = vi.fn((inputs: readonly { sessionId: string; cwd: string }[]) => Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" }))));
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: () => {
createCalls += 1;
return Promise.resolve(busy.runtime);
},
archiveStore: {
list: () => Promise.resolve([]),
get: () => Promise.resolve(undefined),
archive: (input) => Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" }),
archiveMany,
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([sessionRecord("busy"), sessionRecord("ok")]),
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("busy"));
const result = await service.archiveMany([{ id: "busy", cwd: "/workspace" }, { id: "ok", cwd: "/workspace" }, { id: "missing", cwd: "/workspace" }]);
expect(createCalls).toBe(1);
expect(busy.calls.abort).toBe(0);
expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["ok"]);
expect(result.archivedSessionIds).toEqual(["ok"]);
expect(result.failures).toEqual([
{ sessionId: "busy", error: "Stop current session activity before archiving" },
{ sessionId: "missing", error: "Session not found" },
]);
await service.dispose();
});
it("bulk deletes only archived sessions and skips busy active archived runtimes", async () => {
const busyRecord = { sessionId: "busy-archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/busy.jsonl" };
const idleRecord = { sessionId: "idle-archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/idle.jsonl" };
const busy = fakeRuntime("busy-archived", { isStreaming: true });
const deleteArchivedMany = vi.fn((sessionIds: readonly string[]) => Promise.resolve([...sessionIds]));
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(busy.runtime),
archiveStore: {
list: () => Promise.resolve([busyRecord, idleRecord]),
get: (sessionId) => Promise.resolve(sessionId === "busy-archived" ? busyRecord : undefined),
archive: () => { throw new Error("archive should not be called for records that already have archive files"); },
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
deleteArchived: () => Promise.resolve(),
deleteArchivedMany,
},
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([sessionRecord("unarchived")]),
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("busy-archived"));
const result = await service.deleteArchivedMany([{ id: "busy-archived", cwd: "/workspace" }, { id: "idle-archived", cwd: "/workspace" }, { id: "unarchived", cwd: "/workspace" }]);
expect(busy.calls.abort).toBe(0);
expect(deleteArchivedMany).toHaveBeenCalledWith(["idle-archived"]);
expect(result.deletedSessionIds).toEqual(["idle-archived"]);
expect(result.failures).toEqual([
{ sessionId: "busy-archived", error: "Stop current session activity before deleting archived session" },
{ sessionId: "unarchived", error: "Archived session not found" },
]);
await service.dispose();
});
it("bulk delete moves legacy archived records with one workspace scan before deleting", async () => {
const archiveMany = vi.fn((inputs: readonly { sessionId: string; cwd: string }[]) => Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z", archivePath: `/archive/${input.sessionId}.jsonl` }))));
const deleteArchivedMany = vi.fn((sessionIds: readonly string[]) => Promise.resolve([...sessionIds]));
const listCalls: string[] = [];
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([
{ sessionId: "legacy-a", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z" },
{ sessionId: "legacy-b", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z" },
{ sessionId: "moved", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/moved.jsonl" },
]),
get: () => Promise.resolve(undefined),
archive: (input) => Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-01-03T00:00:00.000Z" }),
archiveMany,
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
deleteArchived: () => Promise.resolve(),
deleteArchivedMany,
},
sessionManager: {
create: () => fakeSessionManager(),
list: (cwd) => {
listCalls.push(cwd);
return Promise.resolve([sessionRecord("legacy-a"), sessionRecord("legacy-b"), sessionRecord("unarchived")]);
},
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
const result = await service.deleteArchivedMany([{ id: "legacy-a", cwd: "/workspace" }, { id: "legacy-b", cwd: "/workspace" }, { id: "moved", cwd: "/workspace" }]);
expect(listCalls).toEqual(["/workspace"]);
expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["legacy-a", "legacy-b"]);
expect(deleteArchivedMany).toHaveBeenCalledWith(["legacy-a", "legacy-b", "moved"]);
expect(result.deletedSessionIds).toEqual(["legacy-a", "legacy-b", "moved"]);
expect(result.failures).toEqual([]);
await service.dispose();
});
it("previews session cleanup without mutating and executes a recomputed plan", async () => {
const archivedInputs: string[] = [];
const deletedSessionIds: string[] = [];
let listAllCalls = 0;
const archived = { sessionId: "archived-old", cwd: "/old-project", archivedAt: "2026-04-01T00:00:00.000Z", archivePath: "/archive/archived-old.jsonl" };
const otherArchived = { sessionId: "archived-other", cwd: "/other-project", archivedAt: "2026-04-01T00:00:00.000Z", archivePath: "/archive/archived-other.jsonl" };
const service = new PiSessionService(new CapturingSessionEventHub(), {
now: () => new Date("2026-06-25T00:00:00.000Z"),
archiveStore: {
list: () => Promise.resolve([archived, otherArchived]),
get: () => Promise.resolve(undefined),
archive: () => Promise.reject(new Error("cleanup should use archiveMany")),
archiveMany: (inputs) => {
archivedInputs.push(...inputs.map((input) => input.sessionId));
return Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-06-25T00:00:00.000Z" })));
},
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
deleteArchived: () => Promise.reject(new Error("cleanup should use deleteArchivedMany")),
deleteArchivedMany: (sessionIds) => {
deletedSessionIds.push(...sessionIds);
return Promise.resolve([...sessionIds]);
},
},
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([]),
listAll: () => {
listAllCalls += 1;
return Promise.resolve([
listAllCalls === 1 ? sessionRecord("preview-only", "/old-project") : sessionRecord("execute-only", "/old-project"),
listAllCalls === 1 ? sessionRecord("preview-other", "/other-project") : sessionRecord("execute-other", "/other-project"),
]);
},
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
const preview = await service.cleanupPreview({ thresholds: { archiveIdleDays: 30, deleteArchivedDays: 30 }, projectCwds: ["/old-project"] });
expect(preview.totals).toEqual({ archiveCount: 1, deleteCount: 1 });
expect(preview.projects).toEqual([{ cwd: "/old-project", archiveCount: 1, deleteCount: 1 }]);
expect(archivedInputs).toEqual([]);
expect(deletedSessionIds).toEqual([]);
const result = await service.cleanup({ thresholds: { archiveIdleDays: 30, deleteArchivedDays: 30 }, projectCwds: ["/old-project"] });
expect(result.archivedSessionIds).toEqual(["execute-only"]);
expect(result.deletedSessionIds).toEqual(["archived-old"]);
expect(archivedInputs).toEqual(["execute-only"]);
expect(deletedSessionIds).toEqual(["archived-old"]);
await service.dispose();
});
it("moves legacy cleanup delete records with one workspace scan before batch deleting", async () => {
const listCalls: string[] = [];
const archiveMany = vi.fn((inputs: readonly { sessionId: string; cwd: string }[]) => Promise.resolve(inputs.map((input) => ({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-06-25T00:00:00.000Z", archivePath: `/archive/${input.sessionId}.jsonl` }))));
const deleteArchivedMany = vi.fn((sessionIds: readonly string[]) => Promise.resolve([...sessionIds]));
const service = new PiSessionService(new CapturingSessionEventHub(), {
now: () => new Date("2026-06-25T00:00:00.000Z"),
archiveStore: {
list: () => Promise.resolve([
{ sessionId: "legacy-a", cwd: "/old-project", archivedAt: "2026-04-01T00:00:00.000Z" },
{ sessionId: "legacy-b", cwd: "/old-project", archivedAt: "2026-04-01T00:00:00.000Z" },
]),
get: () => Promise.resolve(undefined),
archive: () => Promise.reject(new Error("cleanup should use archiveMany")),
archiveMany,
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
deleteArchived: () => Promise.reject(new Error("cleanup should use deleteArchivedMany")),
deleteArchivedMany,
},
sessionManager: {
create: () => fakeSessionManager(),
list: (cwd) => {
listCalls.push(cwd);
return Promise.resolve([sessionRecord("legacy-a", cwd), sessionRecord("legacy-b", cwd)]);
},
listAll: () => Promise.resolve([]),
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
const result = await service.cleanup({ thresholds: { deleteArchivedDays: 30 }, projectCwds: ["/old-project"] });
expect(listCalls).toEqual(["/old-project"]);
expect(archiveMany).toHaveBeenCalledTimes(1);
expect(archiveMany.mock.calls[0]?.[0].map((input) => input.sessionId)).toEqual(["legacy-a", "legacy-b"]);
expect(deleteArchivedMany).toHaveBeenCalledWith(["legacy-a", "legacy-b"]);
expect(result.deletedSessionIds).toEqual(["legacy-a", "legacy-b"]);
await service.dispose();
});
it("skips busy active sessions during cleanup execution", async () => {
const fake = fakeRuntime("busy-open", { isStreaming: true, sessionManager: fakeSessionManager("/old-project"), sessionFile: "/sessions/busy-open.jsonl" });
const archivedInputs: string[] = [];
const service = new PiSessionService(new CapturingSessionEventHub(), {
now: () => new Date("2026-06-25T00:00:00.000Z"),
createAgentRuntime: runtimeCreator(fake.runtime),
archiveStore: {
list: () => Promise.resolve([]),
get: () => Promise.resolve(undefined),
archive: (input) => {
archivedInputs.push(input.sessionId);
return Promise.resolve({ sessionId: input.sessionId, cwd: input.cwd, archivedAt: "2026-06-25T00:00:00.000Z" });
},
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager("/old-project"),
list: () => Promise.resolve([sessionRecord("busy-open", "/old-project")]),
listAll: () => Promise.resolve([sessionRecord("busy-open", "/old-project")]),
open: () => fakeSessionManager("/old-project"),
},
heartbeatIntervalMs: 60_000,
});
await service.status("busy-open");
const result = await service.cleanup({ thresholds: { archiveIdleDays: 1 } });
expect(result.archivedSessionIds).toEqual([]);
expect(result.skippedBusySessionIds).toEqual(["busy-open"]);
expect(archivedInputs).toEqual([]);
expect(fake.calls.abort).toBe(0);
await service.dispose();
});
});
@@ -0,0 +1,109 @@
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, describe, expect, it, vi } from "vitest";
import { createPiWebCustomToolDefinitions, sessionAllowsDelegationTools, type PiSessionManager } from "./piSessionService.js";
import type { SubsessionToolDeps } from "./spawnSubsessionTool.js";
import { fakeSessionManager } from "./piSessionService.testSupport.js";
const tempDirs: string[] = [];
afterEach(async () => {
await Promise.all(tempDirs.splice(0).map((path) => rm(path, { recursive: true, force: true })));
});
function delegationDeps() {
const spawn = vi.fn(() => Promise.resolve({ sessionId: "independent-1", cwd: "/workspace" }));
const subsessions: SubsessionToolDeps = {
spawn: vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/workspace" })),
list: vi.fn(() => Promise.resolve([])),
check: vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/workspace", status: "idle" as const, finalText: "", messageCount: 0 })),
read: vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/workspace", status: "idle" as const, entries: [], total: 0, matched: 0, start: 0, hasMore: false })),
};
return { spawn, subsessions };
}
function toolNames(definitions: ReturnType<typeof createPiWebCustomToolDefinitions>): string[] {
return definitions.map((definition) => definition.name);
}
function manager(id: string, file: string | undefined, entries: readonly unknown[] = []): PiSessionManager {
return fakeSessionManager("/workspace", {
getSessionId: () => id,
getSessionFile: () => file,
getEntries: () => entries,
});
}
describe("delegation tool capability boundary", () => {
it("provides every globally enabled delegation tool to unrestricted sessions", () => {
const { spawn, subsessions } = delegationDeps();
expect(toolNames(createPiWebCustomToolDefinitions("/workspace", true, spawn, subsessions))).toEqual([
"edit",
"spawn_session",
"spawn_subsession",
"list_subsessions",
"check_subsession",
"read_subsession",
]);
});
it("continues to honor global delegation feature flags for unrestricted sessions", () => {
const { spawn } = delegationDeps();
expect(toolNames(createPiWebCustomToolDefinitions("/workspace", true, spawn))).toEqual(["edit", "spawn_session"]);
expect(toolNames(createPiWebCustomToolDefinitions("/workspace", true))).toEqual(["edit"]);
});
it("removes every delegation tool but retains ordinary tools for restricted tracked children", () => {
const { spawn, subsessions } = delegationDeps();
expect(toolNames(createPiWebCustomToolDefinitions("/workspace", false, spawn, subsessions))).toEqual(["edit"]);
});
it.each(["human-created", "spawn_session-created"])("allows delegation for a %s session without tracked-child provenance", async () => {
const sessionManager = manager("session-1", undefined);
const open = vi.fn(() => { throw new Error("no parent session should be opened"); });
await expect(sessionAllowsDelegationTools(sessionManager, { open })).resolves.toBe(true);
expect(open).not.toHaveBeenCalled();
});
it("removes delegation when persisted records verify exact tracked-child provenance", async () => {
const dir = await mkdtemp(join(tmpdir(), "pi-web-delegation-provenance-"));
tempDirs.push(dir);
const parentFile = join(dir, "parent.jsonl");
const childFile = join(dir, "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", parentSession: parentFile })}\n`, "utf8");
const childManager = manager("child-1", childFile, [
{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } },
]);
const parentManager = manager("parent-1", parentFile, [
{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: childFile, cwd: "/workspace" } },
]);
await expect(sessionAllowsDelegationTools(childManager, { open: () => parentManager })).resolves.toBe(false);
});
it("does not treat a copied child marker as tracked provenance without an exact reciprocal file link", async () => {
const dir = await mkdtemp(join(tmpdir(), "pi-web-delegation-copy-"));
tempDirs.push(dir);
const parentFile = join(dir, "parent.jsonl");
const originalChildFile = join(dir, "original-child.jsonl");
const copiedChildFile = join(dir, "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(copiedChildFile, `${JSON.stringify({ type: "session", version: 3, id: "child-1", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/workspace", parentSession: parentFile })}\n`, "utf8");
const copiedChildManager = manager("child-1", copiedChildFile, [
{ type: "custom", customType: "pi-web.subsession.spawned", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1" } },
]);
const parentManager = manager("parent-1", parentFile, [
{ type: "custom", customType: "pi-web.subsession.link", data: { version: 1, spawnedBySessionId: "parent-1", spawnedSessionId: "child-1", spawnedSessionFile: originalChildFile, cwd: "/workspace" } },
]);
await expect(sessionAllowsDelegationTools(copiedChildManager, { open: () => parentManager })).resolves.toBe(true);
});
});
@@ -0,0 +1,546 @@
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { describe, expect, it, vi } from "vitest";
import { PiSessionService, type PiAgentSession, type PiSessionRuntime } from "./piSessionService.js";
import { CapturingSessionEventHub, emptyArchiveStore, fakeRuntime, fakeSessionManager, runtimeCreator, sessionGateway, sessionRecord, sessionRef, type RuntimeCreator } from "./piSessionService.testSupport.js";
function deferred<T = void>() {
let resolve!: (value: T | PromiseLike<T>) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((promiseResolve, promiseReject) => {
resolve = promiseResolve;
reject = promiseReject;
});
return { promise, resolve, reject };
}
describe("PiSessionService lifecycle, listing, and reload", () => {
it("starts sessions through an injected runtime creator", async () => {
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime();
let createCalls = 0;
const createAgentRuntime: RuntimeCreator = async () => {
createCalls += 1;
await Promise.resolve();
return fake.runtime;
};
const service = new PiSessionService(hub, {
createAgentRuntime,
sessionManager: sessionGateway([]),
heartbeatIntervalMs: 60_000,
});
const session = await service.start("/workspace");
expect(createCalls).toBe(1);
expect(fake.calls.bindExtensions).toHaveLength(1);
expect(session).toMatchObject({ id: "session-1", cwd: "/workspace", messageCount: 0 });
expect(service.activeCount()).toBe(1);
expect(hub.globalEvents.some((event) => event.type === "status.update" && event.status.sessionId === "session-1")).toBe(true);
expect(hub.globalEvents.some((event) => event.type === "session.created" && event.session.id === "session-1" && event.session.cwd === "/workspace")).toBe(true);
await service.dispose();
expect(fake.calls.abort).toBe(1);
expect(fake.calls.dispose).toBe(1);
});
it("reports persistence from actual session-file existence for fresh active sessions", async () => {
const dir = await mkdtemp(join(tmpdir(), "pi-web-persisted-"));
const sessionFile = join(dir, "new-session.jsonl");
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("new-session", { sessionFile });
let service: PiSessionService | undefined;
try {
service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([]),
heartbeatIntervalMs: 60_000,
});
const session = await service.start("/workspace");
const createdEvent = hub.globalEvents.find((event) => event.type === "session.created");
expect(session).toMatchObject({ id: "new-session", path: sessionFile, persisted: false });
expect(createdEvent).toMatchObject({ type: "session.created", session: { id: "new-session", persisted: false } });
await expect(service.status(sessionRef("new-session"))).resolves.toMatchObject({ sessionId: "new-session", persisted: false });
await writeFile(sessionFile, '{"type":"session","id":"new-session"}\n', "utf8");
await expect(service.status(sessionRef("new-session"))).resolves.toMatchObject({ sessionId: "new-session", persisted: true });
} finally {
await service?.dispose();
await rm(dir, { recursive: true, force: true });
}
});
it("opens legacy id-only lookups from the default session store gateway", async () => {
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("legacy-session");
const open = vi.fn(() => fakeSessionManager());
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([]),
listAll: () => Promise.resolve([sessionRecord("legacy-session")]),
open,
},
heartbeatIntervalMs: 60_000,
});
await expect(service.status("legacy")).resolves.toMatchObject({ sessionId: "legacy-session" });
expect(open).toHaveBeenCalledWith("/sessions/legacy-session.jsonl");
await service.dispose();
});
it("shares one runtime when concurrent cold lookups resolve to the same session", async () => {
const sessionId = "single-flight-session";
const createStarted = deferred();
const releaseCreate = deferred();
const winnerUnsubscribe = vi.fn();
const loserUnsubscribe = vi.fn();
const winnerSubscribe = vi.fn(() => winnerUnsubscribe);
const loserSubscribe = vi.fn(() => loserUnsubscribe);
const winner = fakeRuntime(sessionId, {
sessionManager: fakeSessionManager("/workspace", {
getSessionId: () => sessionId,
getBranch: () => [{ type: "message", message: { role: "user", content: "shared runtime" } }],
}),
subscribe: winnerSubscribe,
});
const loser = fakeRuntime(sessionId, {
sessionManager: fakeSessionManager("/workspace", { getSessionId: () => sessionId }),
subscribe: loserSubscribe,
});
const runtimes = [winner.runtime, loser.runtime];
let createCalls = 0;
const createAgentRuntime: RuntimeCreator = async () => {
const runtime = runtimes[createCalls];
createCalls += 1;
createStarted.resolve();
await releaseCreate.promise;
if (runtime === undefined) throw new Error("unexpected runtime creation");
return runtime;
};
const gateway = sessionGateway([sessionRecord(sessionId)]);
const open = vi.spyOn(gateway, "open");
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: emptyArchiveStore(),
createAgentRuntime,
sessionManager: gateway,
heartbeatIntervalMs: 60_000,
});
const messagesPromise = service.messages(sessionRef(sessionId));
await createStarted.promise;
const statusPromise = service.status(sessionRef("single-flight"));
await new Promise<void>((resolve) => setImmediate(resolve));
const callsWhileOpening = createCalls;
releaseCreate.resolve();
const [messages, status] = await Promise.all([messagesPromise, statusPromise]);
const activeCount = service.activeCount();
await service.dispose();
expect(callsWhileOpening).toBe(1);
expect(createCalls).toBe(1);
expect(open).toHaveBeenCalledOnce();
expect(activeCount).toBe(1);
expect(messages).toEqual([{ role: "user", content: "shared runtime" }]);
expect(status).toMatchObject({ sessionId });
expect(winnerSubscribe).toHaveBeenCalledOnce();
expect(winnerUnsubscribe).toHaveBeenCalledOnce();
expect(winner.calls.dispose).toBe(1);
expect(loserSubscribe).not.toHaveBeenCalled();
expect(loserUnsubscribe).not.toHaveBeenCalled();
expect(loser.calls.dispose).toBe(0);
});
it("clears a failed pending open so the session can be retried", async () => {
const sessionId = "retry-open-session";
const bindStarted = deferred();
const bindResult = deferred();
const openingError = new Error("extension binding failed");
const failed = fakeRuntime(sessionId, {
bindExtensions: () => {
bindStarted.resolve();
return bindResult.promise;
},
});
const retried = fakeRuntime(sessionId);
const runtimes = [failed.runtime, retried.runtime];
let createCalls = 0;
const createAgentRuntime: RuntimeCreator = () => {
const runtime = runtimes[createCalls];
createCalls += 1;
return runtime === undefined
? Promise.reject(new Error("unexpected runtime creation"))
: Promise.resolve(runtime);
};
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: emptyArchiveStore(),
createAgentRuntime,
sessionManager: sessionGateway([sessionRecord(sessionId)]),
heartbeatIntervalMs: 60_000,
});
const messagesPromise = service.messages(sessionRef(sessionId));
await bindStarted.promise;
const statusPromise = service.status(sessionRef("retry-open"));
await new Promise<void>((resolve) => setImmediate(resolve));
const callsWhileOpening = createCalls;
const failedLookups = Promise.allSettled([messagesPromise, statusPromise]);
bindResult.reject(openingError);
const outcomes = await failedLookups;
expect(callsWhileOpening).toBe(1);
expect(outcomes).toHaveLength(2);
for (const outcome of outcomes) {
expect(outcome.status).toBe("rejected");
if (outcome.status === "rejected") expect(outcome.reason).toBe(openingError);
}
expect(service.activeCount()).toBe(0);
expect(failed.calls.abort).toBe(1);
expect(failed.calls.dispose).toBe(1);
await expect(service.status(sessionRef(sessionId))).resolves.toMatchObject({ sessionId });
expect(createCalls).toBe(2);
expect(service.activeCount()).toBe(1);
await service.dispose();
expect(retried.calls.dispose).toBe(1);
});
it("waits for an in-flight open before disposing the service", async () => {
const sessionId = "dispose-opening-session";
const createStarted = deferred();
const runtimeResult = deferred<PiSessionRuntime>();
const fake = fakeRuntime(sessionId);
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: emptyArchiveStore(),
createAgentRuntime: () => {
createStarted.resolve();
return runtimeResult.promise;
},
sessionManager: sessionGateway([sessionRecord(sessionId)]),
heartbeatIntervalMs: 60_000,
});
const statusPromise = service.status(sessionRef(sessionId));
await createStarted.promise;
let disposeSettled = false;
const disposePromise = service.dispose().then(() => { disposeSettled = true; });
await new Promise<void>((resolve) => setImmediate(resolve));
const settledWhileOpening = disposeSettled;
runtimeResult.resolve(fake.runtime);
await expect(statusPromise).resolves.toMatchObject({ sessionId });
await disposePromise;
expect(settledWhileOpening).toBe(false);
expect(service.activeCount()).toBe(0);
expect(fake.calls.abort).toBe(1);
expect(fake.calls.dispose).toBe(1);
});
it("binds extensions again when the SDK runtime replaces the active session", async () => {
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("session-1");
const replacement = fakeRuntime("session-2");
let rebindSession: ((session: PiAgentSession) => Promise<void>) | undefined;
fake.runtime.setRebindSession = (callback) => { rebindSession = callback; };
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([]),
heartbeatIntervalMs: 60_000,
});
await service.start("/workspace");
Object.defineProperty(fake.runtime, "session", { configurable: true, value: replacement.session });
await rebindSession?.(replacement.session);
expect(fake.calls.bindExtensions).toHaveLength(1);
expect(replacement.calls.bindExtensions).toHaveLength(1);
expect(service.activeCount()).toBe(1);
expect(await service.status("session-2")).toMatchObject({ sessionId: "session-2" });
await service.dispose();
});
it("publishes extension errors reported while binding session extensions", async () => {
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("extension-session", {
bindExtensions: (bindings) => {
bindings.onError?.({ extensionPath: "pi-mcp-adapter", event: "session_start", error: "MCP failed" });
return Promise.resolve();
},
});
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([]),
heartbeatIntervalMs: 60_000,
});
await service.start("/workspace");
expect(hub.sessionEvents).toContainEqual({
sessionId: "extension-session",
event: { type: "session.error", message: "pi-mcp-adapter: MCP failed" },
});
const extensionErrorActivity = hub.globalEvents.find((event) => event.type === "activity.update" && event.activity.sessionId === "extension-session");
expect(extensionErrorActivity).toMatchObject({
type: "activity.update",
activity: { sessionId: "extension-session", phase: "error", label: "extension error", detail: "pi-mcp-adapter: MCP failed" },
});
await service.dispose();
});
it("clears stale active activity once a previously active session becomes idle", async () => {
vi.useFakeTimers();
let service: PiSessionService | undefined;
try {
const hub = new CapturingSessionEventHub();
let listener: ((event: unknown) => void) | undefined;
const fake = fakeRuntime("idle-session", {
isStreaming: true,
subscribe: (next) => {
listener = next;
return () => undefined;
},
});
service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("idle-session")]),
heartbeatIntervalMs: 1_000,
});
await service.status(sessionRef("idle-session"));
hub.globalEvents.length = 0;
listener?.({ type: "agent_start" });
const activityPhases = () => hub.globalEvents
.filter((event) => event.type === "activity.update")
.map((event) => event.activity.phase);
expect(activityPhases()).toEqual(["active"]);
fake.session.isStreaming = false;
await vi.advanceTimersByTimeAsync(1_000);
await vi.advanceTimersByTimeAsync(1_000);
expect(activityPhases()).toEqual(["active", "idle"]);
} finally {
await service?.dispose();
vi.useRealTimers();
}
});
it("publishes idle activity for SDK completion events", async () => {
const hub = new CapturingSessionEventHub();
let listener: ((event: unknown) => void) | undefined;
const fake = fakeRuntime("completion-session", {
subscribe: (next) => {
listener = next;
return () => undefined;
},
});
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("completion-session")]),
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("completion-session"));
hub.globalEvents.length = 0;
listener?.({ type: "tool_execution_end", toolName: "read", isError: false });
expect(hub.globalEvents.filter((event) => event.type === "activity.update")).toMatchObject([
{ activity: { sessionId: "completion-session", phase: "idle", label: "tool complete", detail: "read" } },
]);
await service.dispose();
});
it("uses injected archive and session-manager gateways for listing", async () => {
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([{ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-01T00:00:00.000Z" }]),
get: () => Promise.resolve(undefined),
archive: () => Promise.resolve({ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-01T00:00:00.000Z" }),
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([
{ ...sessionRecord("active"), messageCount: 1, firstMessage: "hello", allMessagesText: "hello" },
{ ...sessionRecord("archived"), messageCount: 2, firstMessage: "bye", allMessagesText: "bye" },
]),
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
const sessions = await service.list("/workspace");
expect(sessions).toHaveLength(2);
expect(sessions[0]).toMatchObject({ id: "active", persisted: true });
expect(sessions[0]?.archived).toBeUndefined();
expect(sessions[1]).toMatchObject({ id: "archived", archived: true, archivedAt: "2026-01-01T00:00:00.000Z" });
await service.dispose();
});
it("lists archived records that have been moved out of the active session directory", async () => {
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([{ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", originalPath: "/sessions/archived.jsonl", archivePath: "/archive/archived.jsonl", created: "2026-01-01T00:00:00.000Z", modified: "2026-01-01T00:01:00.000Z", messageCount: 2, firstMessage: "bye" }]),
get: () => Promise.resolve(undefined),
archive: () => { throw new Error("archive should not be called for moved records"); },
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([{ ...sessionRecord("active"), messageCount: 1, firstMessage: "hello", allMessagesText: "hello" }]),
open: () => fakeSessionManager(),
},
heartbeatIntervalMs: 60_000,
});
const sessions = await service.list("/workspace");
expect(sessions).toHaveLength(2);
expect(sessions[0]).toMatchObject({ id: "active" });
expect(sessions[0]?.archived).toBeUndefined();
expect(sessions[1]).toMatchObject({ id: "archived", path: "/sessions/archived.jsonl", archived: true, archivedAt: "2026-01-02T00:00:00.000Z" });
await service.dispose();
});
it("runs /reload by refreshing the active runtime resources in place", async () => {
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("runtime-reload-session");
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("runtime-reload-session")]),
heartbeatIntervalMs: 60_000,
});
await expect(service.runCommand(sessionRef("runtime-reload-session"), "/reload")).resolves.toEqual({
type: "done",
message: "Session runtime resources reloaded. Extensions, skills, prompt templates, themes, and context/system prompt files are refreshed for this session. Reload the browser page separately for PI WEB browser plugin changes.",
});
expect(fake.calls.reload).toBe(1);
expect(fake.calls.abort).toBe(0);
expect(fake.calls.dispose).toBe(0);
expect(hub.globalEvents.some((event) => event.type === "activity.update" && event.activity.sessionId === "runtime-reload-session" && event.activity.label === "resources reloaded")).toBe(true);
expect(hub.globalEvents.some((event) => event.type === "status.update" && event.status.sessionId === "runtime-reload-session")).toBe(true);
await service.dispose();
});
it("reloads a session by closing the active runtime and re-opening it from disk", async () => {
const first = fakeRuntime("reload-session");
const second = fakeRuntime("reload-session");
const runtimes = [first.runtime, second.runtime];
let createCalls = 0;
const createAgentRuntime: RuntimeCreator = async () => {
await Promise.resolve();
const runtime = runtimes[createCalls];
createCalls += 1;
if (runtime === undefined) throw new Error("unexpected runtime creation");
return runtime;
};
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime,
sessionManager: sessionGateway([sessionRecord("reload-session")]),
heartbeatIntervalMs: 60_000,
});
// Open once so there is an active runtime to reload.
await service.status(sessionRef("reload-session"));
expect(createCalls).toBe(1);
await expect(service.reload(sessionRef("reload-session"))).resolves.toBeUndefined();
// The original runtime was torn down and a fresh one opened from disk.
expect(first.calls.abort).toBe(1);
expect(first.calls.dispose).toBe(1);
expect(createCalls).toBe(2);
expect(service.activeCount()).toBe(1);
await service.dispose();
});
it("refuses to reload a session that has active work in progress", async () => {
const fake = fakeRuntime("busy-session", { isStreaming: true });
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("busy-session")]),
heartbeatIntervalMs: 60_000,
});
await expect(service.reload(sessionRef("busy-session"))).rejects.toThrow("Stop current session activity before reloading");
expect(fake.calls.abort).toBe(0);
expect(fake.calls.dispose).toBe(0);
await service.dispose();
});
it("refuses to reload an archived session", async () => {
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([]),
get: (sessionId) => Promise.resolve(sessionId === "archived" || "archived".startsWith(sessionId)
? { sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", archivePath: "/archive/archived.jsonl" }
: undefined),
archive: () => Promise.resolve({ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z" }),
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(true),
},
sessionManager: sessionGateway([]),
heartbeatIntervalMs: 60_000,
});
await expect(service.reload(sessionRef("archived"))).rejects.toThrow("Archived sessions are read-only");
await service.dispose();
});
it("reconciles workspace activity when listing only archived sessions", async () => {
const reconciliations: { cwd: string; sessionIds: string[] }[] = [];
const service = new PiSessionService(new CapturingSessionEventHub(), {
archiveStore: {
list: () => Promise.resolve([{ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", originalPath: "/sessions/archived.jsonl", archivePath: "/archive/archived.jsonl", created: "2026-01-01T00:00:00.000Z", modified: "2026-01-01T00:01:00.000Z", messageCount: 2, firstMessage: "bye" }]),
get: () => Promise.resolve(undefined),
archive: () => { throw new Error("archive should not be called for moved records"); },
restore: () => Promise.resolve(),
isArchived: () => Promise.resolve(false),
},
sessionManager: {
create: () => fakeSessionManager(),
list: () => Promise.resolve([]),
open: () => fakeSessionManager(),
},
workspaceActivity: {
applySessionStatus: () => undefined,
applySessionActivity: () => undefined,
removeSession: () => undefined,
reconcileSessionActivity: (cwd, sessionIds) => { reconciliations.push({ cwd, sessionIds: [...sessionIds] }); },
},
heartbeatIntervalMs: 60_000,
});
const sessions = await service.list("/workspace");
expect(sessions).toHaveLength(1);
expect(sessions[0]).toMatchObject({ id: "archived", archived: true });
expect(reconciliations).toEqual([{ cwd: "/workspace", sessionIds: [] }]);
await service.dispose();
});
});
@@ -0,0 +1,299 @@
import { createAssistantMessageEventStream, type AssistantMessage } from "@earendil-works/pi-ai";
import type { StreamFn } from "@earendil-works/pi-agent-core";
import { AuthStorage, ModelRegistry } from "@earendil-works/pi-coding-agent";
import { describe, expect, it, vi } from "vitest";
import { PiSessionService } from "./piSessionService.js";
import { CapturingSessionEventHub, fakeRuntime, runtimeCreator, sessionGateway, sessionRecord, sessionRef, TEST_MODEL_ID, TEST_MODEL_PROVIDER, testModel, type RuntimeCreator } from "./piSessionService.testSupport.js";
describe("PiSessionService prompt, queue, and auth warnings", () => {
it("sends prompts to an injected runtime without touching the SDK runtime", async () => {
const fake = fakeRuntime("prompt-session");
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("prompt-session")]),
heartbeatIntervalMs: 60_000,
});
await service.prompt(sessionRef("prompt-session"), "Build the thing");
expect(fake.calls.prompt).toEqual([{ text: "Build the thing", options: undefined }]);
await service.dispose();
});
it("echoes the user message for direct prompts but not command-forwarded ones", async () => {
const fake = fakeRuntime("echo-session", {
resourceLoader: { getSkills: () => ({ skills: [{ name: "skill-creator" }] }) },
});
const hub = new CapturingSessionEventHub();
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("echo-session")]),
heartbeatIntervalMs: 60_000,
});
await service.prompt(sessionRef("echo-session"), "Build the thing");
expect(hub.sessionEvents.filter(({ event }) => event.type === "message.append")).toHaveLength(1);
// The client optimistically renders command-forwarded prompts (e.g. /skill:*),
// so the server must not publish a second copy via message.append.
await service.runCommand(sessionRef("echo-session"), "/skill:skill-creator");
expect(hub.sessionEvents.filter(({ event }) => event.type === "message.append")).toHaveLength(1);
expect(fake.calls.prompt).toEqual([
{ text: "Build the thing", options: undefined },
{ text: "/skill:skill-creator", options: undefined },
]);
await service.dispose();
});
it("rejects malformed prompt text before opening the runtime", async () => {
const fake = fakeRuntime("prompt-session");
let createCalls = 0;
const createAgentRuntime: RuntimeCreator = async () => {
createCalls += 1;
await Promise.resolve();
return fake.runtime;
};
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime,
sessionManager: sessionGateway([sessionRecord("prompt-session")]),
heartbeatIntervalMs: 60_000,
});
await expect(service.prompt("prompt-session", undefined)).rejects.toThrow("Prompt text is required");
expect(createCalls).toBe(0);
expect(fake.calls.prompt).toEqual([]);
await service.dispose();
});
it("generates a session name for the first prompt via the session's agent.streamFn", async () => {
const model = testModel();
const streamCalls: unknown[] = [];
const streamFn: StreamFn = (streamModel, context, options) => {
streamCalls.push({ streamModel, context, options });
const stream = createAssistantMessageEventStream();
const message: AssistantMessage = {
role: "assistant",
content: [{ type: "text", text: "Fix login bug" }],
api: "anthropic-messages",
provider: "anthropic",
model: model.id,
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
stopReason: "stop",
timestamp: Date.now(),
};
stream.push({ type: "done", reason: "stop", message });
stream.end(message);
return stream;
};
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("name-session", { model, agent: { streamFn } });
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("name-session")]),
heartbeatIntervalMs: 60_000,
});
await service.prompt(sessionRef("name-session"), "Please fix the login bug");
await vi.waitFor(() => { expect(fake.session.sessionName).toBe("Fix login bug"); });
expect(streamCalls).toHaveLength(1);
expect(hub.sessionEvents.some(({ event }) => event.type === "session.name" && event.name === "Fix login bug")).toBe(true);
await service.dispose();
});
it("includes queued message details in session status", async () => {
const fake = fakeRuntime("status-session", {
messages: [{ role: "user", content: "hello" }, { role: "assistant", content: "hi" }],
pendingMessageCount: 2,
getSteeringMessages: () => ["adjust this turn"],
getFollowUpMessages: () => ["then do this"],
});
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("status-session")]),
heartbeatIntervalMs: 60_000,
});
await expect(service.status(sessionRef("status-session"))).resolves.toMatchObject({
pendingMessageCount: 2,
queuedMessages: [{ kind: "steer", text: "adjust this turn" }, { kind: "followUp", text: "then do this" }],
messageCount: 2,
});
await service.dispose();
});
it("does not enqueue duplicate queued message text", async () => {
const fake = fakeRuntime("dedupe-session", {
isStreaming: true,
pendingMessageCount: 1,
getFollowUpMessages: () => ["already queued"],
});
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("dedupe-session")]),
heartbeatIntervalMs: 60_000,
});
await service.prompt(sessionRef("dedupe-session"), "already queued", "followUp");
expect(fake.calls.prompt).toEqual([]);
await service.dispose();
});
it("does not append queued prompts to the transcript before delivery", async () => {
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("queued-session", { isStreaming: true });
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("queued-session")]),
heartbeatIntervalMs: 60_000,
});
await service.prompt(sessionRef("queued-session"), "Wait for the current turn", "followUp");
expect(fake.calls.prompt).toEqual([{ text: "Wait for the current turn", options: { streamingBehavior: "followUp" } }]);
expect(hub.sessionEvents.some(({ event }) => event.type === "message.append")).toBe(false);
await service.dispose();
});
it("holds prompts sent during compaction until compaction finishes", async () => {
const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("compacting-session", { isCompacting: true });
let resolveFirstPrompt: (() => void) | undefined;
fake.session.prompt = (text: string, options?: { streamingBehavior?: "steer" | "followUp" }) => {
fake.calls.prompt.push({ text, options });
if (options === undefined) {
fake.session.isStreaming = true;
return new Promise<void>((resolve) => { resolveFirstPrompt = resolve; });
}
return Promise.resolve();
};
const service = new PiSessionService(hub, {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("compacting-session")]),
heartbeatIntervalMs: 60_000,
});
await service.prompt(sessionRef("compacting-session"), "Start task 1", "followUp");
await service.prompt(sessionRef("compacting-session"), "Then task 2", "followUp");
expect(fake.calls.prompt).toEqual([]);
expect(hub.sessionEvents.some(({ event }) => event.type === "message.append")).toBe(false);
await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({
pendingMessageCount: 2,
queuedMessages: [{ kind: "followUp", text: "Start task 1" }, { kind: "followUp", text: "Then task 2" }],
});
fake.session.isCompacting = false;
fake.emit({ type: "compaction_end" });
await new Promise((resolve) => setTimeout(resolve, 5));
expect(fake.calls.prompt).toEqual([{ text: "Start task 1", options: undefined }]);
expect(hub.sessionEvents.some(({ event }) => event.type === "message.append" && JSON.stringify(event.message).includes("Start task 1"))).toBe(true);
await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({
pendingMessageCount: 1,
queuedMessages: [{ kind: "followUp", text: "Then task 2" }],
});
fake.emit({ type: "agent_start" });
await new Promise((resolve) => setTimeout(resolve, 5));
expect(fake.calls.prompt).toEqual([
{ text: "Start task 1", options: undefined },
{ text: "Then task 2", options: { streamingBehavior: "followUp" } },
]);
await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({
pendingMessageCount: 0,
queuedMessages: [],
});
resolveFirstPrompt?.();
await service.dispose();
});
it("clears queued messages when aborting active work", async () => {
const fake = fakeRuntime("abort-session");
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("abort-session")]),
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("abort-session"));
await service.abort(sessionRef("abort-session"));
expect(fake.calls.clearQueue).toBe(1);
expect(fake.calls.abort).toBe(1);
await service.dispose();
});
it("clears prompts queued during compaction when aborting active work", async () => {
const fake = fakeRuntime("abort-compaction-session", { isCompacting: true });
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("abort-compaction-session")]),
heartbeatIntervalMs: 60_000,
});
await service.prompt(sessionRef("abort-compaction-session"), "Do not deliver after abort", "followUp");
await expect(service.status(sessionRef("abort-compaction-session"))).resolves.toMatchObject({ pendingMessageCount: 1 });
await service.abort(sessionRef("abort-compaction-session"));
expect(fake.calls.clearQueue).toBe(1);
expect(fake.calls.prompt).toEqual([]);
await expect(service.status(sessionRef("abort-compaction-session"))).resolves.toMatchObject({ pendingMessageCount: 0, queuedMessages: [] });
await service.dispose();
});
it("refreshes auth state and dedupes warnings when logout removes the current model's credentials", async () => {
const hub = new CapturingSessionEventHub();
const authStorage = AuthStorage.inMemory({ anthropic: { type: "api_key", key: "sk-test" } });
const modelRegistry = ModelRegistry.inMemory(authStorage);
const model = modelRegistry.find(TEST_MODEL_PROVIDER, TEST_MODEL_ID);
if (model === undefined) throw new Error("Expected Anthropic model fixture");
const fake = fakeRuntime("auth-session", { model, modelRegistry });
const service = new PiSessionService(hub, {
modelRegistry,
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("auth-session")]),
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("auth-session"));
hub.sessionEvents.length = 0;
hub.globalEvents.length = 0;
authStorage.logout("anthropic");
service.applyAuthChange({ removedProviderId: "anthropic" });
service.applyAuthChange({ removedProviderId: "anthropic" });
const warningCount = () => hub.sessionEvents.filter(({ event }) => event.type === "command.output" && event.level === "error" && event.message.includes(`${TEST_MODEL_PROVIDER}/${TEST_MODEL_ID}`)).length;
expect(warningCount()).toBe(1);
expect(hub.globalEvents.some((event) => event.type === "status.update" && event.status.sessionId === "auth-session")).toBe(true);
authStorage.set("anthropic", { type: "api_key", key: "sk-new" });
service.applyAuthChange();
authStorage.logout("anthropic");
service.applyAuthChange({ removedProviderId: "anthropic" });
expect(warningCount()).toBe(2);
await service.dispose();
});
it("clears queued messages when stopping a session runtime", async () => {
const fake = fakeRuntime("stop-session");
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([sessionRecord("stop-session")]),
heartbeatIntervalMs: 60_000,
});
await service.status(sessionRef("stop-session"));
service.stop(sessionRef("stop-session"));
expect(fake.calls.clearQueue).toBe(1);
await service.dispose();
});
});
@@ -0,0 +1,88 @@
import { describe, expect, it } from "vitest";
import { PiSessionService, type PiAgentSession } from "./piSessionService.js";
import type { SpawnTargetDecision } from "./spawnTargetResolver.js";
import { CapturingSessionEventHub, fakeRuntime, runtimeCreator, sessionGateway, testModel, type RuntimeCreator } from "./piSessionService.testSupport.js";
describe("PiSessionService", () => {
describe("spawnSession", () => {
function spawnService(decision: SpawnTargetDecision) {
const fake = fakeRuntime("spawned-1", { sessionFile: "/tmp/spawned-1.jsonl" });
const log: { details: Record<string, unknown>; message: string }[] = [];
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([]),
spawnTargets: { resolveSpawnTarget: () => Promise.resolve(decision) },
logger: { info: (details, message) => { log.push({ details, message }); } },
heartbeatIntervalMs: 60_000,
});
return { fake, service, log };
}
it("starts a session at the resolved target, delivers the prompt, and logs the spawn", async () => {
const { fake, service, log } = spawnService({ allowed: true, cwd: "/workspace-feature" });
const result = await service.spawnSession({ spawningCwd: "/workspace", prompt: "continue the plan", cwd: "/workspace-feature" });
expect(result).toEqual({ sessionId: "spawned-1", cwd: "/workspace-feature" });
expect(fake.calls.prompt).toEqual([{ text: "continue the plan", options: undefined }]);
expect(log).toEqual([{ details: { spawningCwd: "/workspace", sessionId: "spawned-1", cwd: "/workspace-feature", promptLength: 17 }, message: "spawn_session started a new session" }]);
await service.dispose();
});
it("uses the dispatching session's model as the spawned session's initial model", async () => {
const fake = fakeRuntime("spawned-1", { sessionFile: "/tmp/spawned-1.jsonl" });
const model = testModel();
let initialModel: PiAgentSession["model"];
let delegationToolsEnabled: boolean | undefined;
const createAgentRuntime: RuntimeCreator = async (_createRuntime, options) => {
await Promise.resolve();
initialModel = options.initialModel;
delegationToolsEnabled = options.delegationToolsEnabled;
return fake.runtime;
};
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime,
sessionManager: sessionGateway([]),
spawnTargets: { resolveSpawnTarget: () => Promise.resolve({ allowed: true, cwd: "/workspace-feature" }) },
heartbeatIntervalMs: 60_000,
});
await service.spawnSession({ spawningCwd: "/workspace", prompt: "continue", cwd: "/workspace-feature", model });
expect(initialModel).toBe(model);
expect(delegationToolsEnabled).toBe(true);
await service.dispose();
});
it("rejects an out-of-project target without starting a session", async () => {
const { fake, service } = spawnService({ allowed: false, reason: "out-of-project", allowedCwds: ["/workspace"] });
await expect(service.spawnSession({ spawningCwd: "/workspace", prompt: "go", cwd: "/elsewhere" }))
.rejects.toThrow("cwd must be a workspace of this project. Allowed: /workspace");
expect(fake.calls.prompt).toEqual([]);
expect(service.activeCount()).toBe(0);
await service.dispose();
});
it("rejects when the spawning session is not in a registered project", async () => {
const { service } = spawnService({ allowed: false, reason: "not-registered" });
await expect(service.spawnSession({ spawningCwd: "/workspace", prompt: "go", cwd: undefined }))
.rejects.toThrow("Spawning session is not in a registered project");
await service.dispose();
});
it("is disabled when no spawn target resolver is configured", async () => {
const fake = fakeRuntime("spawned-x");
const service = new PiSessionService(new CapturingSessionEventHub(), {
createAgentRuntime: runtimeCreator(fake.runtime),
sessionManager: sessionGateway([]),
heartbeatIntervalMs: 60_000,
});
await expect(service.spawnSession({ spawningCwd: "/workspace", prompt: "go", cwd: undefined }))
.rejects.toThrow("Spawning sessions is disabled");
await service.dispose();
});
});
});
@@ -0,0 +1,167 @@
import { AuthStorage, ModelRegistry } from "@earendil-works/pi-coding-agent";
import type { GlobalSessionEvent, SessionUiEvent } from "../../shared/apiTypes.js";
import { SessionEventHub } from "../realtime/sessionEventHub.js";
import type { PiAgentSession, PiSessionManager, PiSessionRuntime, PiSessionServiceDependencies } from "./piSessionService.js";
export class CapturingSessionEventHub extends SessionEventHub {
readonly sessionEvents: { sessionId: string; event: SessionUiEvent }[] = [];
readonly globalEvents: GlobalSessionEvent[] = [];
override publish(sessionId: string, event: SessionUiEvent): void {
this.sessionEvents.push({ sessionId, event });
}
override publishGlobal(event: GlobalSessionEvent): void {
this.globalEvents.push(event);
}
}
export type SessionGateway = NonNullable<PiSessionServiceDependencies["sessionManager"]>;
export type RuntimeCreator = NonNullable<PiSessionServiceDependencies["createAgentRuntime"]>;
export interface TestSession extends PiAgentSession {
sessionName: string | undefined;
model: PiAgentSession["model"];
isStreaming: boolean;
isCompacting: boolean;
isBashRunning: boolean;
pendingMessageCount: number;
getSteeringMessages: () => readonly string[];
getFollowUpMessages: () => readonly string[];
}
export function fakeSessionManager(cwd = "/workspace", patch: Partial<PiSessionManager> = {}): PiSessionManager {
return {
getCwd: () => cwd,
getSessionId: () => "session-1",
getSessionFile: () => undefined,
getBranch: () => [],
getLeafId: () => "leaf-1",
...patch,
};
}
export function sessionRecord(id: string, cwd = "/workspace") {
return { id, path: `/sessions/${id}.jsonl`, cwd, created: new Date("2026-01-01T00:00:00.000Z"), modified: new Date("2026-01-01T00:01:00.000Z"), messageCount: 0, firstMessage: "", allMessagesText: "" };
}
export function sessionRef(id: string, cwd = "/workspace") {
return { id, cwd };
}
export const TEST_MODEL_PROVIDER = "anthropic";
export const TEST_MODEL_ID = "claude-sonnet-4-5-20250929";
export function testModel(): NonNullable<PiAgentSession["model"]> {
const model = ModelRegistry.inMemory(AuthStorage.inMemory()).find(TEST_MODEL_PROVIDER, TEST_MODEL_ID);
if (model === undefined) throw new Error("test model not found");
return model;
}
export function fakeRuntime(sessionId = "session-1", patch: Partial<TestSession> = {}) {
const promptCalls: { text: string; options: unknown }[] = [];
const customMessageCalls: { message: { customType: string; content: string; display: boolean; details?: unknown }; options: unknown }[] = [];
const bindExtensionCalls: unknown[] = [];
const listeners: ((event: unknown) => void)[] = [];
const calls = { abort: 0, bindExtensions: bindExtensionCalls, clearQueue: 0, dispose: 0, prompt: promptCalls, reload: 0, sendCustomMessage: customMessageCalls };
const session: TestSession = {
sessionId,
sessionFile: `/tmp/${sessionId}.jsonl`,
messages: [],
sessionName: undefined,
model: undefined,
thinkingLevel: "off",
isStreaming: false,
isCompacting: false,
isBashRunning: false,
pendingMessageCount: 0,
sessionManager: fakeSessionManager(),
modelRegistry: ModelRegistry.create(AuthStorage.inMemory()),
scopedModels: [],
extensionRunner: { getRegisteredCommands: () => [] },
promptTemplates: [],
resourceLoader: { getSkills: () => ({ skills: [] }) },
subscribe: (listener: (event: unknown) => void) => {
listeners.push(listener);
return () => {
const index = listeners.indexOf(listener);
if (index !== -1) listeners.splice(index, 1);
};
},
bindExtensions: (bindings: unknown) => {
calls.bindExtensions.push(bindings);
return Promise.resolve();
},
getSessionStats: () => ({ sessionId, totalMessages: 0, userMessages: 0, assistantMessages: 0, toolCalls: 0, tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, cost: 0 }),
getContextUsage: () => undefined,
reload: () => {
calls.reload += 1;
return Promise.resolve();
},
prompt: (text: string, options: unknown) => {
calls.prompt.push({ text, options });
return Promise.resolve();
},
sendCustomMessage: (message: { customType: string; content: string; display: boolean; details?: unknown }, options: unknown) => {
calls.sendCustomMessage.push({ message, options });
return Promise.resolve();
},
executeBash: () => Promise.resolve({ output: "", exitCode: 0, cancelled: false, truncated: false }),
abort: () => {
calls.abort += 1;
return Promise.resolve();
},
clearQueue: () => {
calls.clearQueue += 1;
return { steering: [], followUp: [] };
},
getSteeringMessages: () => [],
getFollowUpMessages: () => [],
setModel: () => Promise.resolve(),
cycleModel: () => Promise.resolve(undefined),
getAvailableThinkingLevels: () => [],
setThinkingLevel: () => undefined,
cycleThinkingLevel: () => undefined,
setSessionName: (name: string) => { session.sessionName = name; },
compact: () => Promise.resolve({ summary: "", tokensBefore: 0 }),
getUserMessagesForForking: () => [],
agent: { streamFn: () => { throw new Error("streamFn should not be called in this test"); } },
...patch,
};
const runtime: PiSessionRuntime = {
cwd: session.sessionManager.getCwd(),
session,
setRebindSession: () => undefined,
fork: () => Promise.resolve({ cancelled: false }),
dispose: () => {
calls.dispose += 1;
return Promise.resolve();
},
};
return { runtime, session, calls, emit: (event: unknown) => { for (const listener of [...listeners]) listener(event); } };
}
export function runtimeCreator(runtime: PiSessionRuntime): RuntimeCreator {
return async () => {
await Promise.resolve();
return runtime;
};
}
export function sessionGateway(records: ReturnType<typeof sessionRecord>[]): SessionGateway {
return {
create: () => fakeSessionManager(),
list: () => Promise.resolve(records),
open: () => fakeSessionManager(),
};
}
export 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),
};
}
File diff suppressed because it is too large Load Diff
@@ -1,6 +1,6 @@
import { constants } from "node:fs";
import { access, mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { join, resolve } from "node:path";
import { tmpdir } from "node:os";
import { afterEach, describe, expect, it } from "vitest";
import { SessionArchiveStore } from "./sessionArchiveStore.js";
@@ -71,6 +71,101 @@ describe("SessionArchiveStore", () => {
expect(await exists(record.archivePath)).toBe(false);
await expect(store.list()).resolves.toEqual([]);
});
it("archives and permanently deletes sessions in batches", async () => {
const root = await mkdtemp(join(tmpdir(), "pi-web-archive-batch-"));
tempRoots.push(root);
const activeDir = join(root, "active");
await mkdir(activeDir, { recursive: true });
const sourceA = join(activeDir, "2026-01-01_a.jsonl");
const sourceB = join(activeDir, "2026-01-01_b.jsonl");
await writeFile(sourceA, "a\n", "utf8");
await writeFile(sourceB, "b\n", "utf8");
const store = new SessionArchiveStore(join(root, "archived-sessions.json"), join(root, "archived-files"));
const records = await store.archiveMany([
{
sessionId: "a",
cwd: "/workspace",
path: sourceA,
created: "2026-01-01T00:00:00.000Z",
modified: "2026-01-01T00:01:00.000Z",
messageCount: 1,
firstMessage: "a",
},
{
sessionId: "b",
cwd: "/workspace",
path: sourceB,
created: "2026-01-01T00:00:00.000Z",
modified: "2026-01-01T00:02:00.000Z",
messageCount: 2,
firstMessage: "b",
},
]);
expect(records.map((record) => record.sessionId)).toEqual(["a", "b"]);
expect(await exists(sourceA)).toBe(false);
expect(await exists(sourceB)).toBe(false);
await expect(store.list()).resolves.toMatchObject([{ sessionId: "a" }, { sessionId: "b" }]);
const archivePaths = records.map((record) => record.archivePath);
if (archivePaths.some((path) => path === undefined)) throw new Error("Expected archive paths");
await expect(store.deleteArchivedMany(["a", "b", "missing"])).resolves.toEqual(["a", "b"]);
for (const archivePath of archivePaths) {
if (archivePath === undefined) throw new Error("Expected archive path");
expect(await exists(archivePath)).toBe(false);
}
await expect(store.list()).resolves.toEqual([]);
});
it("prefers exact persisted session IDs over prefix matches and canonicalizes stored cwd", async () => {
const root = await mkdtemp(join(tmpdir(), "pi-web-archive-prefix-"));
tempRoots.push(root);
const archiveFile = join(root, "archived-sessions.json");
const rawCwd = join(root, "workspace", "..", "workspace");
await writeFile(archiveFile, JSON.stringify({
sessions: [
{
sessionId: "abc123",
cwd: rawCwd,
archivedAt: "2026-01-01T00:00:00.000Z",
originalPath: "/sessions/abc123.jsonl",
archivePath: "/archive/abc123.jsonl",
messageCount: 3,
firstMessage: "prefix",
name: "Prefix match",
parentSessionPath: "/sessions/root.jsonl",
},
{
sessionId: "abc",
cwd: rawCwd,
archivedAt: "2026-01-01T00:00:00.000Z",
originalPath: "/sessions/abc.jsonl",
archivePath: "/archive/abc.jsonl",
messageCount: 1,
firstMessage: "exact",
},
],
}), "utf8");
const store = new SessionArchiveStore(archiveFile, join(root, "archived-files"));
await expect(store.get("abc")).resolves.toMatchObject({
sessionId: "abc",
cwd: resolve(rawCwd),
firstMessage: "exact",
});
await expect(store.get("abc1")).resolves.toMatchObject({
sessionId: "abc123",
cwd: resolve(rawCwd),
firstMessage: "prefix",
name: "Prefix match",
parentSessionPath: "/sessions/root.jsonl",
});
await expect(store.isArchived("abc1")).resolves.toBe(true);
await expect(store.isArchived("missing")).resolves.toBe(false);
});
});
async function exists(path: string): Promise<boolean> {
+49 -19
View File
@@ -53,24 +53,39 @@ export class SessionArchiveStore {
}
async archive(session: ArchiveSessionInput): Promise<ArchivedSessionRecord> {
const [record] = await this.archiveMany([session]);
if (record === undefined) throw new Error("Archive operation did not produce a record");
return record;
}
async archiveMany(sessions: readonly ArchiveSessionInput[]): Promise<ArchivedSessionRecord[]> {
if (sessions.length === 0) return [];
return this.exclusive(async () => {
const data = await this.read();
const existingIndex = data.sessions.findIndex((record) => record.sessionId === session.sessionId);
const existing = existingIndex === -1 ? undefined : data.sessions[existingIndex];
const archivePath = existing?.archivePath ?? this.archivePathFor(session);
const record = archiveRecordFromInput(session, {
archivedAt: existing?.archivedAt ?? new Date().toISOString(),
originalPath: existing?.originalPath ?? session.path,
archivePath,
});
const records: ArchivedSessionRecord[] = [];
const filesToRemove: { source: string; archivePath: string }[] = [];
await copySessionFileToArchive(session.path, archivePath);
for (const session of sessions) {
const existingIndex = data.sessions.findIndex((record) => record.sessionId === session.sessionId);
const existing = existingIndex === -1 ? undefined : data.sessions[existingIndex];
const archivePath = existing?.archivePath ?? this.archivePathFor(session);
const record = archiveRecordFromInput(session, {
archivedAt: existing?.archivedAt ?? new Date().toISOString(),
originalPath: existing?.originalPath ?? session.path,
archivePath,
});
await copySessionFileToArchive(session.path, archivePath);
if (existingIndex === -1) data.sessions.push(record);
else data.sessions[existingIndex] = record;
records.push(record);
filesToRemove.push({ source: session.path, archivePath });
}
if (existingIndex === -1) data.sessions.push(record);
else data.sessions[existingIndex] = record;
await this.write(data);
await removeActiveSessionFile(session.path, archivePath);
return record;
for (const file of filesToRemove) await removeActiveSessionFile(file.source, file.archivePath);
return records;
});
}
@@ -90,14 +105,25 @@ export class SessionArchiveStore {
}
async deleteArchived(sessionId: string): Promise<void> {
await this.exclusive(async () => {
const data = await this.read();
const record = data.sessions.find((session) => session.sessionId === sessionId);
if (record === undefined) return;
await this.deleteArchivedMany([sessionId]);
}
if (record.archivePath !== undefined && await pathExists(record.archivePath)) await unlink(record.archivePath);
const sessions = data.sessions.filter((session) => session.sessionId !== sessionId);
async deleteArchivedMany(sessionIds: readonly string[]): Promise<string[]> {
const targetIds = uniqueStrings(sessionIds);
if (targetIds.length === 0) return [];
return this.exclusive(async () => {
const data = await this.read();
const targetIdSet = new Set(targetIds);
const records = data.sessions.filter((session) => targetIdSet.has(session.sessionId));
if (records.length === 0) return [];
for (const record of records) {
if (record.archivePath !== undefined && await pathExists(record.archivePath)) await unlink(record.archivePath);
}
const sessions = data.sessions.filter((session) => !targetIdSet.has(session.sessionId));
await this.write({ sessions });
const deletedIds = new Set(records.map((record) => record.sessionId));
return targetIds.filter((sessionId) => deletedIds.has(sessionId));
});
}
@@ -254,6 +280,10 @@ function safeFileName(value: string): string {
return value.replace(/[^a-zA-Z0-9._-]/g, "_") || "session";
}
function uniqueStrings(values: readonly string[]): string[] {
return [...new Set(values)];
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null;
}
@@ -11,11 +11,11 @@ function candidate(id: string, options: Partial<SessionArchiveTreeCandidate> = {
}
describe("session archive tree planning", () => {
it("finds candidates by full id or prefix", () => {
const candidates = [candidate("abcdef"), candidate("xyz")];
it("finds candidates by exact id before falling back to a prefix", () => {
const candidates = [candidate("abcdef"), candidate("abc"), candidate("xyz")];
expect(findArchiveCandidateByIdOrPrefix(candidates, "abcdef")?.id).toBe("abcdef");
expect(findArchiveCandidateByIdOrPrefix(candidates, "abc")?.id).toBe("abcdef");
expect(findArchiveCandidateByIdOrPrefix(candidates, "abc")?.id).toBe("abc");
expect(findArchiveCandidateByIdOrPrefix(candidates, "abcd")?.id).toBe("abcdef");
expect(findArchiveCandidateByIdOrPrefix(candidates, "missing")).toBeUndefined();
});
@@ -66,11 +66,15 @@ describe("SessionCommandService", () => {
await expect(service.run("s1", "/template arg")).resolves.toMatchObject({ type: "done" });
await expect(service.run("s1", "/skill:skill-a arg")).resolves.toMatchObject({ type: "done" });
expect(prompt).toHaveBeenCalledTimes(3);
expect(prompt).toHaveBeenNthCalledWith(1, "s1", "/ext arg");
expect(prompt).toHaveBeenNthCalledWith(2, "s1", "/template arg");
expect(prompt).toHaveBeenNthCalledWith(3, "s1", "/skill:skill-a arg");
});
it("renames sessions and returns updated client session metadata", async () => {
it("renames sessions, publishes the name update, and returns updated client session metadata", async () => {
const active = activeSession();
const service = new SessionCommandService(() => getActive(active), vi.fn(), eventPublisher());
const events = eventPublisher();
const service = new SessionCommandService(() => getActive(active), vi.fn(), events);
await expect(service.run("s1", "/name Useful name")).resolves.toMatchObject({
type: "done",
@@ -78,6 +82,7 @@ describe("SessionCommandService", () => {
session: { id: "s1", cwd: "/work", name: "Useful name", messageCount: 2 },
});
expect(active.runtime.session.setSessionName).toHaveBeenCalledWith("Useful name");
expect(events.publish).toHaveBeenCalledWith("s1", { type: "session.name", sessionId: "s1", name: "Useful name" });
});
it("formats session stats", async () => {
@@ -90,22 +95,50 @@ describe("SessionCommandService", () => {
});
});
it("starts compaction and publishes completion", async () => {
it("starts compaction, updates lifecycle hooks, and publishes completion", async () => {
const active = activeSession();
const events = eventPublisher();
const service = new SessionCommandService(() => getActive(active), vi.fn(), events);
const onCompactionStart = vi.fn();
const onCompactionEnd = vi.fn();
const service = new SessionCommandService(() => getActive(active), vi.fn(), events, { onCompactionStart, onCompactionEnd });
await expect(service.run("s1", "/compact focus on tests")).resolves.toEqual({ type: "done", message: "Compaction started…" });
expect(onCompactionStart).toHaveBeenCalledWith(active.runtime.session);
await vi.waitFor(() => {
expect(events.publish).toHaveBeenCalledWith("s1", {
type: "command.output",
level: "success",
message: "Compaction complete.\nTokens before: 123\n\nshort summary",
});
expect(onCompactionEnd).toHaveBeenCalledWith(active.runtime.session, "success");
});
expect(active.runtime.session.compact).toHaveBeenCalledWith("focus on tests");
});
it("reloads runtime resources through the injected lifecycle callback", async () => {
const active = activeSession();
const reloadSession = vi.fn(async () => { await Promise.resolve(); });
const service = new SessionCommandService(() => getActive(active), vi.fn(), eventPublisher(), { reloadSession });
await expect(service.run("s1", "/reload")).resolves.toEqual({
type: "done",
message: "Session runtime resources reloaded. Extensions, skills, prompt templates, themes, and context/system prompt files are refreshed for this session. Reload the browser page separately for PI WEB browser plugin changes.",
});
expect(reloadSession).toHaveBeenCalledWith(active.runtime.session);
});
it("rejects runtime reload while the session has active work", async () => {
const active = activeSession({ isBashRunning: true });
const reloadSession = vi.fn(async () => { await Promise.resolve(); });
const service = new SessionCommandService(() => getActive(active), vi.fn(), eventPublisher(), { reloadSession });
await expect(service.run("s1", "/reload")).resolves.toEqual({
type: "unsupported",
message: "Cannot reload while the session is active. Stop current activity before reloading.",
});
expect(reloadSession).not.toHaveBeenCalled();
});
it("creates fork selection requests from newest message to oldest and responds with selected entry", async () => {
const active = activeSession({
getUserMessagesForForking: vi.fn(() => [
+21 -4
View File
@@ -47,6 +47,12 @@ export interface CommandEventPublisher {
publishGlobal?(event: Extract<SessionUiEvent, { type: "session.name" }>): void;
}
export interface SessionCommandLifecycle<TSession extends CommandSession = CommandSession> {
onCompactionStart?: (session: TSession) => void;
onCompactionEnd?: (session: TSession, result: "success" | "error", detail?: string) => void;
reloadSession?: (session: TSession) => Promise<void>;
}
export interface SessionCommandNaming {
listSessionNames?: (cwd: string) => Promise<readonly string[]>;
}
@@ -65,10 +71,7 @@ export class SessionCommandService<TSession extends CommandSession = CommandSess
private readonly getActive: GetCommandActiveSession<TSession>,
private readonly prompt: (sessionId: string, text: string) => Promise<void>,
private readonly events: CommandEventPublisher,
private readonly lifecycle: {
onCompactionStart?: (session: TSession) => void;
onCompactionEnd?: (session: TSession, result: "success" | "error", detail?: string) => void;
} = {},
private readonly lifecycle: SessionCommandLifecycle<TSession> = {},
private readonly naming: SessionCommandNaming = {},
) {}
@@ -93,6 +96,7 @@ export class SessionCommandService<TSession extends CommandSession = CommandSess
if (name === "session") return { type: "done", message: formatSessionStats(session) };
if (name === "name") return this.nameSession(active, rest);
if (name === "compact") return this.compact(session, rest);
if (name === "reload") return this.reload(session);
if (name === "clone") return this.clone(active);
if (name === "fork") return this.fork(active);
@@ -140,6 +144,19 @@ export class SessionCommandService<TSession extends CommandSession = CommandSess
return { type: "done", message: "Compaction started…" };
}
private async reload(session: TSession): Promise<ClientCommandResult> {
if (sessionHasActiveWork(session)) return { type: "unsupported", message: "Cannot reload while the session is active. Stop current activity before reloading." };
if (this.lifecycle.reloadSession === undefined) return { type: "unsupported", message: "/reload is not available for this session runtime." };
try {
await this.lifecycle.reloadSession(session);
} catch (error: unknown) {
const message = error instanceof Error ? error.message : String(error);
return { type: "unsupported", message: `Reload failed: ${message}` };
}
return { type: "done", message: "Session runtime resources reloaded. Extensions, skills, prompt templates, themes, and context/system prompt files are refreshed for this session. Reload the browser page separately for PI WEB browser plugin changes." };
}
private async clone(active: CommandActiveSession<TSession>): Promise<ClientCommandResult> {
if (sessionHasActiveWork(active.runtime.session)) return forkActiveUnsupported("clone");
const leafId = active.runtime.session.sessionManager.getLeafId();
@@ -1,11 +1,89 @@
import type { Api, AssistantMessage, Model } from "@earendil-works/pi-ai";
import { createAssistantMessageEventStream } from "@earendil-works/pi-ai";
import type { StreamFn } from "@earendil-works/pi-agent-core";
import { describe, expect, it } from "vitest";
import { cleanSessionName, fallbackSessionName } from "./sessionNameGenerator.js";
import { cleanSessionName, deterministicSessionName, fallbackSessionName, generateShortSessionName } from "./sessionNameGenerator.js";
function fakeModel(): Model<Api> {
return { id: "fake-model", name: "Fake Model", api: "anthropic-messages", provider: "anthropic", baseUrl: "https://example.test", reasoning: false, input: ["text"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 1000, maxTokens: 100 };
}
function fakeAssistantMessage(overrides: Partial<AssistantMessage> = {}): AssistantMessage {
return {
role: "assistant",
content: [],
api: "anthropic-messages",
provider: "anthropic",
model: "fake-model",
usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
stopReason: "stop",
timestamp: Date.now(),
...overrides,
};
}
function streamThatCompletes(text: string): StreamFn {
return () => {
const stream = createAssistantMessageEventStream();
const message = fakeAssistantMessage({ content: [{ type: "text", text }] });
stream.push({ type: "done", reason: "stop", message });
stream.end(message);
return stream;
};
}
function streamThatErrors(): StreamFn {
return () => {
const stream = createAssistantMessageEventStream();
const message = fakeAssistantMessage({ stopReason: "error", errorMessage: "boom" });
stream.push({ type: "error", reason: "error", error: message });
stream.end(message);
return stream;
};
}
describe("sessionNameGenerator", () => {
it("generates a session name by calling the injected streamFn", async () => {
const calls: unknown[] = [];
const stream = streamThatCompletes('Title: "Fix the bug"');
const streamFn: StreamFn = (model, context, options) => {
calls.push({ model, context, options });
return stream(model, context, options);
};
const name = await generateShortSessionName(streamFn, fakeModel(), "Please fix the login bug");
expect(name).toBe("Fix the bug");
expect(calls).toHaveLength(1);
});
it("returns undefined when the stream reports an error", async () => {
const streamFn = streamThatErrors();
const name = await generateShortSessionName(streamFn, fakeModel(), "Please fix the login bug");
expect(name).toBeUndefined();
});
it("cleans model-generated titles", () => {
expect(cleanSessionName('Title: "Fix Session Naming."\nextra')).toBe("Fix Session Naming");
});
it("builds deterministic names for relay handoff prompts", () => {
expect(deterministicSessionName('Relay "handoff-check" leg 2 begins now.\n\nYou are the next runner.'))
.toBe("Relay handoff-check leg 2");
});
it("preserves the relay leg when truncating deterministic relay names", () => {
expect(deterministicSessionName('Relay "very-long-relay-name-that-would-otherwise-push-the-leg-number-out-of-view" leg 42 begins now.'))
.toBe("Relay very-long-relay-name-that-would-otherwise-push leg 42");
});
it("does not build deterministic names for non-canonical relay prompts", () => {
expect(deterministicSessionName('You are continuing Relay "handoff-check" under the Relay method.'))
.toBeUndefined();
});
it("builds a concise fallback from the first request", () => {
expect(fallbackSessionName("Seems like auto name for sessions is not working, I still get the first message as a name."))
.toBe("Seems like auto name for sessions");
+28 -54
View File
@@ -1,33 +1,20 @@
import type { Api, AssistantMessage, AssistantMessageEventStream, Context, Model, SimpleStreamOptions } from "@earendil-works/pi-ai";
import type { ModelRegistry } from "@earendil-works/pi-coding-agent";
import type { Api, AssistantMessage, Model } from "@earendil-works/pi-ai";
import type { StreamFn } from "@earendil-works/pi-agent-core";
const SESSION_NAME_TIMEOUT_MS = 10_000;
const SESSION_NAME_MAX_INPUT_CHARS = 4_000;
const SESSION_NAME_MAX_LENGTH = 60;
const FALLBACK_SESSION_NAME_MAX_WORDS = 6;
const PI_AI_COMPAT_MODULE = ["@earendil-works/pi-ai", "compat"].join("/");
const RELAY_HANDOFF_FIRST_LINE = /^Relay\s+"([^"\n]+)"\s+leg\s+(\d+)\s+begins now\.?\s*(?:\n|$)/;
interface SessionNameApiProvider {
streamSimple(model: Model<Api>, context: Context, options?: SimpleStreamOptions): AssistantMessageEventStream;
export function deterministicSessionName(firstMessage: unknown): string | undefined {
if (typeof firstMessage !== "string") return undefined;
return relayHandoffSessionName(firstMessage.trimStart());
}
interface PiAiProviderRegistryModule {
getApiProvider?: (api: Api) => SessionNameApiProvider | undefined;
}
type ModuleImporter = (specifier: string) => Promise<unknown>;
let piAiProviderRegistryModulePromise: Promise<PiAiProviderRegistryModule> | undefined;
export async function generateShortSessionName<TApi extends Api>(modelRegistry: ModelRegistry, model: Model<TApi>, firstMessage: string): Promise<string | undefined> {
const providerRegistry = await getPiAiProviderRegistryModule();
const provider = providerRegistry.getApiProvider?.(model.api);
if (provider === undefined) return undefined;
const auth = await modelRegistry.getApiKeyAndHeaders(model);
if (!auth.ok) return undefined;
const stream = provider.streamSimple(
export async function generateShortSessionName<TApi extends Api>(streamFn: StreamFn, model: Model<TApi>, firstMessage: string): Promise<string | undefined> {
const stream = await streamFn(
model,
{
systemPrompt: "Generate a concise title for a coding-agent chat session. Return only the title, with no quotes or punctuation wrapper.",
@@ -41,8 +28,6 @@ export async function generateShortSessionName<TApi extends Api>(modelRegistry:
maxTokens: 24,
reasoning: "minimal",
signal: AbortSignal.timeout(SESSION_NAME_TIMEOUT_MS),
...(auth.apiKey === undefined ? {} : { apiKey: auth.apiKey }),
...(auth.headers === undefined ? {} : { headers: auth.headers }),
},
);
@@ -81,40 +66,29 @@ export function cleanSessionName(value: string): string | undefined {
return title === "" ? undefined : title;
}
async function getPiAiProviderRegistryModule(importer: ModuleImporter = (specifier) => import(specifier)): Promise<PiAiProviderRegistryModule> {
piAiProviderRegistryModulePromise ??= loadPiAiProviderRegistryModule(importer);
return piAiProviderRegistryModulePromise;
function relayHandoffSessionName(firstMessage: string): string | undefined {
const match = RELAY_HANDOFF_FIRST_LINE.exec(firstMessage);
if (match === null) return undefined;
const relayName = match[1]?.replace(/\s+/g, " ").trim();
const legNumber = match[2];
if (relayName === undefined || relayName === "" || legNumber === undefined) return undefined;
return cleanSessionName(formatRelaySessionName(relayName, legNumber));
}
async function loadPiAiProviderRegistryModule(importer: ModuleImporter): Promise<PiAiProviderRegistryModule> {
const compatModule = await importOptionalPiAiModule(PI_AI_COMPAT_MODULE, importer);
if (hasGetApiProvider(compatModule)) return compatModule;
const rootModule = await importer("@earendil-works/pi-ai");
if (hasGetApiProvider(rootModule)) return rootModule;
return {};
function formatRelaySessionName(relayName: string, legNumber: string): string {
const prefix = "Relay ";
const suffix = ` leg ${legNumber}`;
const maxRelayNameLength = Math.max(1, SESSION_NAME_MAX_LENGTH - prefix.length - suffix.length);
const displayedRelayName = truncateRelayName(relayName, maxRelayNameLength);
return `${prefix}${displayedRelayName}${suffix}`;
}
async function importOptionalPiAiModule(specifier: string, importer: ModuleImporter): Promise<unknown> {
try {
return await importer(specifier);
} catch (error) {
if (isModuleUnavailableError(error)) return undefined;
throw error;
}
}
function hasGetApiProvider(moduleValue: unknown): moduleValue is PiAiProviderRegistryModule {
return typeof moduleValue === "object"
&& moduleValue !== null
&& "getApiProvider" in moduleValue
&& typeof moduleValue.getApiProvider === "function";
}
function isModuleUnavailableError(error: unknown): boolean {
if (!(error instanceof Error)) return false;
const code = "code" in error ? error.code : undefined;
return code === "ERR_MODULE_NOT_FOUND" || code === "ERR_PACKAGE_PATH_NOT_EXPORTED";
function truncateRelayName(relayName: string, maxLength: number): string {
if (relayName.length <= maxLength) return relayName;
const truncated = relayName.slice(0, maxLength).replace(/[\s._-]+$/g, "").trim();
return truncated === "" ? relayName.slice(0, maxLength).trim() : truncated;
}
function textFromAssistant(message: AssistantMessage): string {
+86 -2
View File
@@ -2,7 +2,7 @@ import { resolve } from "node:path";
import Fastify, { type FastifyInstance } from "fastify";
import fastifyWebsocket from "@fastify/websocket";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import type { SessionCleanupExecuteResponse, SessionCleanupPreviewResponse } from "../../shared/apiTypes.js";
import type { MessagePage, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkMutationRef, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse } from "../../shared/apiTypes.js";
import { SessionEventHub } from "../realtime/sessionEventHub.js";
import { PiSessionService, type PiSessionManagerGateway } from "./piSessionService.js";
import type { SessionRouteLookup, SessionRouteService } from "./sessionService.js";
@@ -56,6 +56,32 @@ describe("session routes", () => {
}
});
it("omits thinking signatures from browser history without mutating service messages", async () => {
const routeApp = Fastify({ logger: false });
await routeApp.register(fastifyWebsocket);
const eventHub = new SessionEventHub();
const routeService = new CapturingRouteSessionService();
const thinkingBlock = { type: "thinking", thinking: "private chain", thinkingSignature: "opaque-provider-payload", redacted: true };
const message = { role: "assistant", content: [thinkingBlock, { type: "text", text: "visible answer" }] };
routeService.messagesResponse = { messages: [message], start: 0, total: 1 };
registerSessionRoutes(routeApp, routeService, eventHub);
try {
const response = await routeApp.inject({ method: "GET", url: "/sessions/session-1/messages?limit=20" });
expect(response.statusCode).toBe(200);
expect(response.json()).toEqual({
messages: [{ role: "assistant", content: [{ type: "thinking", thinking: "private chain", redacted: true }, { type: "text", text: "visible answer" }] }],
start: 0,
total: 1,
});
expect(thinkingBlock.thinkingSignature).toBe("opaque-provider-payload");
} finally {
await routeService.dispose();
await routeApp.close();
}
});
it("forwards prompt attachments and supports the save-attachments route", async () => {
const routeApp = Fastify({ logger: false });
await routeApp.register(fastifyWebsocket);
@@ -179,13 +205,59 @@ describe("session routes", () => {
await routeApp.close();
}
});
it("routes bulk archive and delete requests with normalized session refs", async () => {
const routeApp = Fastify({ logger: false });
await routeApp.register(fastifyWebsocket);
const eventHub = new SessionEventHub();
const routeService = new CapturingRouteSessionService();
registerSessionRoutes(routeApp, routeService, eventHub);
try {
const requestCwd = resolve("/repo");
const archiveResponse = await routeApp.inject({ method: "POST", url: "/sessions/bulk/archive", payload: { sessions: [{ id: "s1", cwd: requestCwd }, { id: "s2" }] } });
const deleteResponse = await routeApp.inject({ method: "POST", url: "/sessions/bulk/delete-archived", payload: { sessions: [{ id: "s1", cwd: requestCwd }] } });
expect(archiveResponse.statusCode).toBe(200);
expect(archiveResponse.json()).toMatchObject({ archived: true, archivedSessionIds: ["s1", "s2"], failures: [] });
expect(deleteResponse.statusCode).toBe(200);
expect(deleteResponse.json()).toMatchObject({ deleted: true, deletedSessionIds: ["s1"], failures: [] });
expect(routeService.bulkArchiveCalls).toEqual([[{ id: "s1", cwd: requestCwd }, { id: "s2" }]]);
expect(routeService.bulkDeleteCalls).toEqual([[{ id: "s1", cwd: requestCwd }]]);
} finally {
await routeService.dispose();
await routeApp.close();
}
});
it("rejects malformed bulk mutation bodies before calling the service", async () => {
const routeApp = Fastify({ logger: false });
await routeApp.register(fastifyWebsocket);
const eventHub = new SessionEventHub();
const routeService = new CapturingRouteSessionService();
registerSessionRoutes(routeApp, routeService, eventHub);
try {
const response = await routeApp.inject({ method: "POST", url: "/sessions/bulk/archive", payload: { sessions: [{ cwd: "/repo" }] } });
expect(response.statusCode).toBe(400);
expect(response.json()).toEqual({ error: "id field must be a string" });
expect(routeService.bulkArchiveCalls).toEqual([]);
} finally {
await routeService.dispose();
await routeApp.close();
}
});
});
class CapturingRouteSessionService implements SessionRouteService {
readonly calls: unknown[] = [];
readonly reloadCalls: SessionRouteLookup[] = [];
messagesResponse: unknown[] | MessagePage = [];
readonly cleanupPreviewCalls: NormalizedSessionCleanupRequest[] = [];
readonly cleanupCalls: NormalizedSessionCleanupRequest[] = [];
readonly bulkArchiveCalls: SessionBulkMutationRef[][] = [];
readonly bulkDeleteCalls: SessionBulkMutationRef[][] = [];
reloadError: Error | undefined;
cleanupPreview(request: NormalizedSessionCleanupRequest): Promise<SessionCleanupPreviewResponse> {
@@ -198,6 +270,16 @@ class CapturingRouteSessionService implements SessionRouteService {
return Promise.resolve({ generatedAt: "2026-06-25T00:00:00.000Z", thresholds: request.thresholds, projects: [], totals: { archiveCount: 0, deleteCount: 0 }, archivedSessionIds: [], deletedSessionIds: [] });
}
archiveMany(refs: readonly SessionBulkMutationRef[]): Promise<SessionBulkArchiveResponse> {
this.bulkArchiveCalls.push([...refs]);
return Promise.resolve({ archived: true, archivedSessionIds: refs.map((ref) => ref.id), failures: [], generatedAt: "2026-06-25T00:00:00.000Z" });
}
deleteArchivedMany(refs: readonly SessionBulkMutationRef[]): Promise<SessionBulkDeleteArchivedResponse> {
this.bulkDeleteCalls.push([...refs]);
return Promise.resolve({ deleted: true, deletedSessionIds: refs.map((ref) => ref.id), failures: [], generatedAt: "2026-06-25T00:00:00.000Z" });
}
reload(lookup: SessionRouteLookup): Promise<void> {
this.reloadCalls.push(lookup);
if (this.reloadError !== undefined) return Promise.reject(this.reloadError);
@@ -210,7 +292,9 @@ class CapturingRouteSessionService implements SessionRouteService {
list(): never { throw unusedRouteMethod("list"); }
start(): never { throw unusedRouteMethod("start"); }
messages(): Promise<unknown[]> { return Promise.resolve([]); }
messages(): Promise<unknown[] | MessagePage> {
return Promise.resolve(this.messagesResponse);
}
status(lookup: SessionRouteLookup) {
this.calls.push(lookup);
+37 -2
View File
@@ -1,5 +1,6 @@
import type { FastifyInstance } from "fastify";
import type { SessionCleanupRequest } from "../../shared/apiTypes.js";
import type { SessionBulkMutationRequest, SessionBulkMutationRef, SessionCleanupRequest } from "../../shared/apiTypes.js";
import { projectBrowserMessageResponse } from "../browserMessageProjection.js";
import { normalizeRequestCwd } from "../workingDirectory.js";
import type { SessionEventHub } from "../realtime/sessionEventHub.js";
import type { SessionRouteLookup, SessionRouteService } from "./sessionService.js";
@@ -64,10 +65,27 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: SessionRou
}
});
app.post<{ Body: SessionBulkMutationRequest | undefined }>(`${prefix}/sessions/bulk/archive`, async (request, reply) => {
try {
return await sessions.archiveMany(bulkMutationRefsFromBody(request.body));
} catch (error) {
return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
}
});
app.post<{ Body: SessionBulkMutationRequest | undefined }>(`${prefix}/sessions/bulk/delete-archived`, async (request, reply) => {
try {
return await sessions.deleteArchivedMany(bulkMutationRefsFromBody(request.body));
} catch (error) {
return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
}
});
app.get<{ Params: { sessionId: string }; Querystring: MessageQuery }>(`${prefix}/sessions/:sessionId/messages`, async (request, reply) => {
try {
const page = { ...optionalField("before", optionalNumber(request.query.before)), ...optionalField("limit", optionalNumber(request.query.limit)) };
return await sessions.messages(sessionLookupFromQuery(request.params.sessionId, request.query), page);
const messages = await sessions.messages(sessionLookupFromQuery(request.params.sessionId, request.query), page);
return projectBrowserMessageResponse(messages);
} catch (error) {
return reply.code(404).send({ error: errorMessage(error) });
}
@@ -281,6 +299,23 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: SessionRou
});
}
function bulkMutationRefsFromBody(body: SessionBulkMutationRequest | undefined): SessionBulkMutationRef[] {
const record = requireRecord(body);
const sessions = record["sessions"];
if (!Array.isArray(sessions)) throw new Error("sessions field must be an array");
return sessions.map(parseBulkMutationRef);
}
function parseBulkMutationRef(value: unknown): SessionBulkMutationRef {
const record = requireRecord(value);
const id = requireString(record, "id").trim();
if (id === "") throw new Error("id field must not be empty");
const cwd = record["cwd"];
if (cwd === undefined || cwd === "") return { id };
if (typeof cwd !== "string") throw new Error("cwd field must be a string");
return { id, cwd: normalizeRequestCwd(cwd) };
}
function sessionLookupFromQuery(id: string, query: SessionQuery): SessionLookup {
return sessionLookupFromCwd(id, query.cwd);
}
+8 -1
View File
@@ -1,4 +1,9 @@
import type { SavedPromptAttachment } from "../../shared/apiTypes.js";
import type {
SavedPromptAttachment,
SessionBulkArchiveResponse,
SessionBulkDeleteArchivedResponse,
SessionBulkMutationRef,
} from "../../shared/apiTypes.js";
import type {
ClientArchiveSessionsResponse,
ClientCommand,
@@ -40,6 +45,8 @@ export interface SessionRouteService {
saveAttachments(ref: SessionRouteLookup, attachments: unknown, folder?: string): Promise<SavedPromptAttachment[]>;
cleanupPreview(request: NormalizedSessionCleanupRequest): Promise<ClientSessionCleanupPreviewResponse>;
cleanup(request: NormalizedSessionCleanupRequest): Promise<ClientSessionCleanupExecuteResponse>;
archiveMany(refs: readonly SessionBulkMutationRef[]): Promise<SessionBulkArchiveResponse>;
deleteArchivedMany(refs: readonly SessionBulkMutationRef[]): Promise<SessionBulkDeleteArchivedResponse>;
shell(ref: SessionRouteLookup, text: string): Promise<void>;
runCommand(ref: SessionRouteLookup, text: string): Promise<ClientCommandResult>;
respondToCommand(ref: SessionRouteLookup, requestId: string, value: string): Promise<ClientCommandResult>;
+17 -8
View File
@@ -2,23 +2,32 @@ import type { ExtensionContext } from "@earendil-works/pi-coding-agent";
import { describe, expect, it, vi } from "vitest";
import { createSpawnSessionToolDefinition } from "./spawnSessionTool.js";
// The spawn tool's execute() never reads ctx, so an empty stub is sufficient.
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions -- test stub; execute() does not use ctx.
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions -- test stub with the minimal surface the tool reads.
const ctx = {} as ExtensionContext;
const dispatchModel = { provider: "anthropic", id: "claude-sonnet" };
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions -- test stub with the minimal surface the tool reads.
const ctxWithModel = { model: dispatchModel } as ExtensionContext;
describe("createSpawnSessionToolDefinition", () => {
it("passes the spawning cwd and params to the spawn callback and reports success", async () => {
it("passes the spawning cwd, explicit cwd, dispatching model, and prompt to spawn callback", async () => {
const spawn = vi.fn(() => Promise.resolve({ sessionId: "new-1", cwd: "/repos/a-feature" }));
const tool = createSpawnSessionToolDefinition("/repos/a", { spawn });
const result = await tool.execute("call-1", { prompt: "do the thing", cwd: "/repos/a-feature" }, undefined, undefined, ctx);
const result = await tool.execute("call-1", { prompt: "do the thing", cwd: "/repos/a-feature" }, undefined, undefined, ctxWithModel);
expect(spawn).toHaveBeenCalledWith({ spawningCwd: "/repos/a", prompt: "do the thing", cwd: "/repos/a-feature" });
expect(spawn).toHaveBeenCalledWith({ spawningCwd: "/repos/a", prompt: "do the thing", cwd: "/repos/a-feature", model: dispatchModel });
expect(result.details).toEqual({ sessionId: "new-1", cwd: "/repos/a-feature" });
expect(result.content[0]).toMatchObject({ type: "text", text: "Started session new-1 in /repos/a-feature." });
expect(result.content[0]).toMatchObject({ type: "text", text: "Started independent session new-1 in /repos/a-feature." });
});
it("defaults cwd to undefined so the service falls back to the spawning cwd", async () => {
it("describes the independent-session capability without workflow policy", () => {
const tool = createSpawnSessionToolDefinition("/repos/a", { spawn: vi.fn() });
expect(tool.description).toBe("Start a new independent pi-web session and send it an initial prompt. The session is not tracked by the caller, can be opened by a human, and runs without returning its later output to the caller.");
expect(tool.description).not.toMatch(/use this|continue work|follow a plan|relay/i);
});
it("forwards omitted cwd as undefined and omits a missing dispatching model", async () => {
const spawn = vi.fn(() => Promise.resolve({ sessionId: "new-2", cwd: "/repos/a" }));
const tool = createSpawnSessionToolDefinition("/repos/a", { spawn });
@@ -31,7 +40,7 @@ describe("createSpawnSessionToolDefinition", () => {
const spawn = vi.fn(() => Promise.reject(new Error("cwd must be a workspace of this project. Allowed: /repos/a")));
const tool = createSpawnSessionToolDefinition("/repos/a", { spawn });
await expect(tool.execute("call-3", { prompt: "x", cwd: "/elsewhere" }, undefined, undefined, ctx))
await expect(tool.execute("call-4", { prompt: "x", cwd: "/elsewhere" }, undefined, undefined, ctx))
.rejects.toThrow("cwd must be a workspace of this project. Allowed: /repos/a");
});
});
+14 -5
View File
@@ -1,15 +1,19 @@
import { Type } from "typebox";
import { defineTool } from "@earendil-works/pi-coding-agent";
import { defineTool, type ExtensionContext } from "@earendil-works/pi-coding-agent";
export interface SpawnSessionResult {
sessionId: string;
cwd: string;
}
export type SpawnSessionModel = NonNullable<ExtensionContext["model"]>;
export interface SpawnSessionInvocation {
spawningCwd: string;
prompt: string;
cwd: string | undefined;
/** Current model from the dispatching session, used as the spawned session's default. */
model?: SpawnSessionModel;
}
export interface SpawnSessionToolDeps {
@@ -37,16 +41,21 @@ export function createSpawnSessionToolDefinition(spawningCwd: string, deps: Spaw
return defineTool<typeof SpawnSessionParams, SpawnSessionToolDetails>({
name: "spawn_session",
label: "Spawn session",
description: "Start a new, independent pi-web session and send it an initial prompt. Use this to dispatch a fresh agent to continue work or follow a plan. The new session runs on its own and a human can interact with it; you do not receive its output.",
description: "Start a new independent pi-web session and send it an initial prompt. The session is not tracked by the caller, can be opened by a human, and runs without returning its later output to the caller.",
promptSnippet: "spawn_session: start a new independent session with a first prompt",
parameters: SpawnSessionParams,
async execute(_toolCallId, params) {
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
// Failures throw: the agent loop turns the thrown message into an error
// tool result the model sees, so the spawning agent can adapt (e.g. pick a
// valid workspace) rather than crash.
const result = await deps.spawn({ spawningCwd, prompt: params.prompt, cwd: params.cwd });
const result = await deps.spawn({
spawningCwd,
prompt: params.prompt,
cwd: params.cwd,
...(ctx.model === undefined ? {} : { model: ctx.model }),
});
return {
content: [{ type: "text", text: `Started session ${result.sessionId} in ${result.cwd}.` }],
content: [{ type: "text", text: `Started independent session ${result.sessionId} in ${result.cwd}.` }],
details: result,
};
},
@@ -3,11 +3,13 @@ import type { ExtensionContext } from "@earendil-works/pi-coding-agent";
import { describe, expect, it, vi } from "vitest";
import { createSubsessionToolDefinitions, type SubsessionToolDeps } from "./spawnSubsessionTool.js";
function ctxFor(sessionId: string, sessionFile: string | undefined): ExtensionContext {
const dispatchModel = { provider: "anthropic", id: "claude-sonnet" };
function ctxFor(sessionId: string, sessionFile: string | undefined, model?: unknown): ExtensionContext {
const sessionManager = { getSessionId: () => sessionId, getSessionFile: () => sessionFile };
// The subsession tools only read sessionManager.getSessionId/getSessionFile.
// The subsession tools only read sessionManager.getSessionId/getSessionFile and model.
// eslint-disable-next-line @typescript-eslint/consistent-type-assertions -- test stub with the minimal surface the tools use.
return { sessionManager } as unknown as ExtensionContext;
return { sessionManager, ...(model === undefined ? {} : { model }) } as unknown as ExtensionContext;
}
function tools(deps: Partial<SubsessionToolDeps>) {
@@ -36,7 +38,7 @@ describe("createSubsessionToolDefinitions", () => {
const spawn = vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/repos/a-feature" }));
const { spawn: spawnTool } = tools({ spawn });
const result = await spawnTool.execute("call-1", { prompt: "do it", cwd: "/repos/a-feature" }, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl"));
const result = await spawnTool.execute("call-1", { prompt: "do it", cwd: "/repos/a-feature" }, undefined, undefined, ctxFor("parent-1", "/sessions/parent-1.jsonl", dispatchModel));
expect(spawn).toHaveBeenCalledWith({
spawningCwd: "/repos/a",
@@ -44,9 +46,48 @@ describe("createSubsessionToolDefinitions", () => {
parentSessionFile: "/sessions/parent-1.jsonl",
prompt: "do it",
cwd: "/repos/a-feature",
model: dispatchModel,
});
expect(result.details).toEqual({ sessionId: "child-1", cwd: "/repos/a-feature" });
expect(firstText(result.content)).toContain("Started subsession child-1");
expect(firstText(result.content)).toContain("Started tracked subsession child-1");
});
it("guides the parent to join all required subsessions without polling", async () => {
const { spawn: spawnTool } = tools({
spawn: vi.fn(() => Promise.resolve({ sessionId: "child-1", cwd: "/repos/a-feature" })),
});
expect(spawnTool.description).toBe("Start a tracked child and return after dispatch. Track required children as pending: continue independent work, then yield at a join point until all have notified completion. Notifications queue while the parent is busy; do not poll for completion.");
expect(spawnTool.promptSnippet).toBe("spawn_subsession: delegate parallel work; yield at a join point until all required children complete.");
const result = await spawnTool.execute("call-contract", { prompt: "do it" }, undefined, undefined, ctxFor("parent-1", undefined));
expect(firstText(result.content)).toBe("Started tracked subsession child-1 in /repos/a-feature. Track it as pending and, before finalizing dependent work, yield until all required children have notified completion.");
});
it("keeps subsession inspection tool descriptions capability-oriented", () => {
const definitions = tools({});
expect(definitions.list.description).toBe("List tracked child sessions owned by the calling session, with each child's current status (working, idle, error, or unknown).");
expect(definitions.check.description).toBe("Return a tracked subsession's current status, message count, and most recent assistant output.");
expect(definitions.read.description).toBe("Return a filtered, paginated transcript of a tracked subsession. Filters select message roles and content kinds, search full message content, optionally include raw tool arguments, and cap or page the returned entries.");
for (const definition of [definitions.list, definitions.check, definitions.read]) {
expect(definition.description).not.toMatch(/use this|do not poll|continue working|start narrow|for just the final|relay/i);
}
});
it("spawn_subsession omits the inherited model when the dispatching session has no current model", async () => {
const spawn = vi.fn(() => Promise.resolve({ sessionId: "child-2", cwd: "/repos/a" }));
const { spawn: spawnTool } = tools({ spawn });
await spawnTool.execute("call-modeless", { prompt: "do it" }, undefined, undefined, ctxFor("parent-1", undefined));
expect(spawn).toHaveBeenCalledWith({
spawningCwd: "/repos/a",
parentSessionId: "parent-1",
parentSessionFile: undefined,
prompt: "do it",
cwd: undefined,
});
});
it("list_subsessions reports the caller's subsessions and their status", async () => {
@@ -69,7 +110,7 @@ describe("createSubsessionToolDefinitions", () => {
it("list_subsessions reports an empty state", async () => {
const { list: listTool } = tools({ list: vi.fn(() => Promise.resolve([])) });
const result = await listTool.execute("call-3", {}, undefined, undefined, ctxFor("parent-1", undefined));
expect(result.content[0]).toMatchObject({ type: "text", text: "You have not spawned any subsessions." });
expect(result.content[0]).toMatchObject({ type: "text", text: "No tracked subsessions." });
});
it("check_subsession scopes by parent and returns the final result", async () => {
+27 -16
View File
@@ -1,5 +1,5 @@
import { Type } from "typebox";
import { defineTool } from "@earendil-works/pi-coding-agent";
import { defineTool, type ExtensionContext } from "@earendil-works/pi-coding-agent";
import type { TranscriptContentKind, TranscriptEntry, TranscriptRole, TranscriptView } from "./subsessionTranscript.js";
/** Lifecycle phase of a tracked subsession as seen by its parent. */
@@ -10,6 +10,8 @@ export interface SpawnSubsessionResult {
cwd: string;
}
export type SpawnSubsessionModel = NonNullable<ExtensionContext["model"]>;
export interface SpawnSubsessionInvocation {
/** cwd of the session that invoked the tool (used for project-scope checks). */
spawningCwd: string;
@@ -19,6 +21,8 @@ export interface SpawnSubsessionInvocation {
parentSessionFile: string | undefined;
prompt: string;
cwd: string | undefined;
/** Current model from the dispatching session, used as the spawned session's default. */
model?: SpawnSubsessionModel;
}
export interface SubsessionSummary {
@@ -74,13 +78,13 @@ const ListSubsessionsParams = Type.Object({});
const CheckSubsessionParams = Type.Object({
sessionId: Type.String({
description: "Id of a subsession you spawned (as returned by spawn_subsession or list_subsessions).",
description: "Id of a tracked subsession owned by the calling session, as returned by spawn_subsession or list_subsessions.",
}),
});
const ReadSubsessionParams = Type.Object({
sessionId: Type.String({
description: "Id of a subsession you spawned (as returned by spawn_subsession or list_subsessions).",
description: "Id of a tracked subsession owned by the calling session, as returned by spawn_subsession or list_subsessions.",
}),
roles: Type.Optional(Type.Array(
Type.Union([Type.Literal("assistant"), Type.Literal("user"), Type.Literal("tool"), Type.Literal("system"), Type.Literal("custom")]),
@@ -122,7 +126,7 @@ function renderEntry(entry: TranscriptEntry): string {
function clipNotice(part: TranscriptEntry["parts"][number]): string {
if ((part.kind === "text" || part.kind === "thinking" || part.kind === "tool_result") && part.truncated !== undefined) {
return ` [+${String(part.truncated.full - part.truncated.shown)} chars truncated; re-read with a larger maxChars]`;
return ` [+${String(part.truncated.full - part.truncated.shown)} chars truncated]`;
}
return "";
}
@@ -149,15 +153,15 @@ function renderTranscript(result: SubsessionReadResult): string {
? "no messages matched your filters"
: `no messages in this window (${String(result.matched)} matched outside it)`)
: `messages ${String(result.start)}${String(last.index)} of ${String(result.total)} (${String(result.matched)} matched)`;
const more = result.hasMore ? `\n\nMore matching messages exist earlier; page back with before: ${String(result.start)}.` : "";
const more = result.hasMore ? `\n\nEarlier matching messages exist before index ${String(result.start)}.` : "";
// Empty entries with matches means the `before` cursor excluded every match
// (they all sit at index >= before): the agent paged too far back and should
// raise `before` or omit it, not page back further.
const body = result.entries.length > 0
? result.entries.map(renderEntry).join("\n\n")
: (result.matched === 0
? "(nothing matched; try widening roles/include, dropping search, or raising limit)"
: `(no messages before index ${String(result.start)}; all ${String(result.matched)} matches are later — raise 'before' or omit it)`);
? "(no messages matched the filters)"
: `(no messages before index ${String(result.start)}; all ${String(result.matched)} matches have later indexes)`);
return `Subsession ${result.sessionId} [${result.status}] — ${range}:\n\n${body}${more}`;
}
@@ -174,15 +178,22 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
const spawnTool = defineTool<typeof SpawnSubsessionParams, SpawnSubsessionResult>({
name: "spawn_subsession",
label: "Spawn subsession",
description: "Start a tracked child session and send it an initial prompt. The subsession runs independently and a human can interact with it, but unlike spawn_session it is linked to you: you are notified when it stops working (finished, idle, or errored), and you can inspect it with list_subsessions, check_subsession (a quick glance at its latest output), and read_subsession (read through its transcript). Use this to delegate work you intend to follow up on.",
promptSnippet: "spawn_subsession: start a tracked child session you will be notified about",
description: "Start a tracked child and return after dispatch. Track required children as pending: continue independent work, then yield at a join point until all have notified completion. Notifications queue while the parent is busy; do not poll for completion.",
promptSnippet: "spawn_subsession: delegate parallel work; yield at a join point until all required children complete.",
parameters: SpawnSubsessionParams,
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
const parentSessionId = ctx.sessionManager.getSessionId();
const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
const result = await deps.spawn({ spawningCwd, parentSessionId, parentSessionFile, prompt: params.prompt, cwd: params.cwd });
const result = await deps.spawn({
spawningCwd,
parentSessionId,
parentSessionFile,
prompt: params.prompt,
cwd: params.cwd,
...(ctx.model === undefined ? {} : { model: ctx.model }),
});
return {
content: [{ type: "text", text: `Started subsession ${result.sessionId} in ${result.cwd}. You will be notified when it stops working.` }],
content: [{ type: "text", text: `Started tracked subsession ${result.sessionId} in ${result.cwd}. Track it as pending and, before finalizing dependent work, yield until all required children have notified completion.` }],
details: result,
};
},
@@ -191,7 +202,7 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
const listTool = defineTool<typeof ListSubsessionsParams, { subsessions: SubsessionSummary[] }>({
name: "list_subsessions",
label: "List subsessions",
description: "List the tracked subsessions you spawned, with their current status (working, idle, error, or unknown).",
description: "List tracked child sessions owned by the calling session, with each child's current status (working, idle, error, or unknown).",
promptSnippet: "list_subsessions: see the tracked child sessions you spawned",
parameters: ListSubsessionsParams,
async execute(_toolCallId, _params, _signal, _onUpdate, ctx) {
@@ -199,8 +210,8 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
const parentSessionFile = ctx.sessionManager.getSessionFile() ?? undefined;
const subsessions = await deps.list(parentSessionId, parentSessionFile);
const text = subsessions.length === 0
? "You have not spawned any subsessions."
: `Your subsessions:\n${subsessions.map(statusLine).join("\n")}`;
? "No tracked subsessions."
: `Tracked subsessions:\n${subsessions.map(statusLine).join("\n")}`;
return { content: [{ type: "text", text }], details: { subsessions } };
},
});
@@ -208,7 +219,7 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
const checkTool = defineTool<typeof CheckSubsessionParams, SubsessionCheckResult>({
name: "check_subsession",
label: "Check subsession",
description: "Quick glance at a subsession you spawned: its current status and most recent assistant output. Use this to react to what a subsession produced. When the summary is not enough, use read_subsession to look through its full transcript.",
description: "Return a tracked subsession's current status, message count, and most recent assistant output.",
promptSnippet: "check_subsession: glance at a subsession's status and latest output",
parameters: CheckSubsessionParams,
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
@@ -226,7 +237,7 @@ export function createSubsessionToolDefinitions(spawningCwd: string, deps: Subse
const readTool = defineTool<typeof ReadSubsessionParams, SubsessionReadResult>({
name: "read_subsession",
label: "Read subsession",
description: "Read through the transcript of a subsession you spawned. Returns its messages filtered and paginated however you ask: choose which roles (assistant, user, tool, system, custom) and content kinds (text, thinking, tool_call, tool_result, image) to include, search for a substring (always over full content), cap each value's length with maxChars (omit for full text; clipped parts are flagged so truncation is never silent), optionally include raw tool args, and page backward with 'before'/'limit'. Start narrow (e.g. assistant text with a small maxChars) and widen the filters, raise maxChars, or page further back if you don't find what you need. For just the final result, use check_subsession instead.",
description: "Return a filtered, paginated transcript of a tracked subsession. Filters select message roles and content kinds, search full message content, optionally include raw tool arguments, and cap or page the returned entries.",
promptSnippet: "read_subsession: read through a subsession's transcript with filters",
parameters: ReadSubsessionParams,
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
@@ -81,12 +81,12 @@ describe("buildTranscriptView", () => {
expect(callPart.args).toEqual({ command: "ls" });
});
it("search keeps only matching entries across text and tool names", () => {
const messages = [assistant("the auth flow"), assistant("unrelated"), toolResult("error in auth.ts", "read")];
it("search keeps only entries matching text or tool-call names", () => {
const messages = [assistant("the auth flow"), assistant("unrelated"), toolResult("error in auth.ts", "read"), toolCall("auth-search")];
const view = buildTranscriptView(messages, { search: "auth" });
expect(view.matched).toBe(2);
expect(view.entries.map((entry) => entry.index)).toEqual([0, 2]);
expect(view.matched).toBe(3);
expect(view.entries.map((entry) => entry.index)).toEqual([0, 2, 3]);
});
it("search runs against full content even when maxChars would clip the match away", () => {