fix: preserve legacy session route lookups

This commit is contained in:
Federico Jaramillo Martinez
2026-06-10 22:42:55 +02:00
parent 71510444c4
commit b99143f757
13 changed files with 350 additions and 129 deletions
+1 -1
View File
@@ -2,4 +2,4 @@
"@jmfederico/pi-web": patch "@jmfederico/pi-web": patch
--- ---
Respect Pi session directory settings in pi-web sessions, including project-local Pi settings, while addressing session operations with their workspace context. Respect Pi session directory settings in pi-web sessions, including project-local Pi settings, while allowing cwd-scoped session operations without breaking legacy id-only routes.
+25 -1
View File
@@ -1,7 +1,7 @@
import { afterEach, describe, expect, it, vi } from "vitest"; import { afterEach, describe, expect, it, vi } from "vitest";
import { PI_WEB_CAPABILITIES } from "../../../shared/capabilities"; import { PI_WEB_CAPABILITIES } from "../../../shared/capabilities";
import type { TerminalCommandRun, Workspace } from "../../../shared/apiTypes"; import type { TerminalCommandRun, Workspace } from "../../../shared/apiTypes";
import { machinesApi, piWebApi, terminalsApi, workspacesApi } from "./clients"; import { machinesApi, piWebApi, sessionsApi, terminalsApi, workspacesApi } from "./clients";
const workspace: Workspace = { const workspace: Workspace = {
id: "w/1", id: "w/1",
@@ -60,6 +60,30 @@ describe("machine-scoped runtime API", () => {
}); });
}); });
describe("session API compatibility", () => {
it("keeps legacy session-id calls free of cwd context", async () => {
const fetchMock = stubJsonFetch({ accepted: true });
await sessionsApi.prompt("s 1", "hello", "followUp", "remote a");
expect(fetchMock).toHaveBeenCalledOnce();
const [url, init] = fetchCall(fetchMock, 0);
expect(url).toBe("/api/machines/remote%20a/sessions/s%201/prompt");
expect(JSON.parse(requestBody(init))).toEqual({ text: "hello", streamingBehavior: "followUp" });
});
it("adds cwd context when session refs include a workspace", async () => {
const fetchMock = stubJsonFetch({ accepted: true });
await sessionsApi.prompt({ id: "s 1", cwd: "/repo" }, "hello", undefined, "remote a");
expect(fetchMock).toHaveBeenCalledOnce();
const [url, init] = fetchCall(fetchMock, 0);
expect(url).toBe("/api/machines/remote%20a/sessions/s%201/prompt");
expect(JSON.parse(requestBody(init))).toEqual({ cwd: "/repo", text: "hello" });
});
});
describe("machine-scoped terminal command-run API", () => { describe("machine-scoped terminal command-run API", () => {
it("deletes workspaces through the selected machine scope", async () => { it("deletes workspaces through the selected machine scope", async () => {
const fetchMock = stubJsonFetch(commandRun); const fetchMock = stubJsonFetch(commandRun);
+49 -26
View File
@@ -42,17 +42,40 @@ import { machineGitDiffUrl, messageUrl } from "./urls";
const machinePrefix = (machineId = "local") => `/api/machines/${encodeURIComponent(machineId)}`; const machinePrefix = (machineId = "local") => `/api/machines/${encodeURIComponent(machineId)}`;
function sessionBaseUrl(session: SessionRef, machineId = "local"): string { type SessionLookup = SessionRef | string;
return `${machinePrefix(machineId)}/sessions/${encodeURIComponent(session.id)}`;
function sessionId(session: SessionLookup): string {
return typeof session === "string" ? session : session.id;
} }
function sessionUrl(session: SessionRef, endpoint: string, machineId = "local"): string { function sessionCwd(session: SessionLookup): string | undefined {
return typeof session === "string" ? undefined : session.cwd;
}
function sessionBaseUrl(session: SessionLookup, machineId = "local"): string {
return `${machinePrefix(machineId)}/sessions/${encodeURIComponent(sessionId(session))}`;
}
function sessionUrl(session: SessionLookup, endpoint: string, machineId = "local"): string {
return `${sessionBaseUrl(session, machineId)}/${endpoint}`; return `${sessionBaseUrl(session, machineId)}/${endpoint}`;
} }
function sessionQueryUrl(session: SessionRef, endpoint: string, machineId = "local"): string { function sessionQueryUrl(session: SessionLookup, endpoint: string, machineId = "local"): string {
const query = new URLSearchParams({ cwd: session.cwd }).toString(); return `${sessionUrl(session, endpoint, machineId)}${sessionQuery(session)}`;
return `${sessionUrl(session, endpoint, machineId)}?${query}`; }
function sessionBaseQueryUrl(session: SessionLookup, machineId = "local"): string {
return `${sessionBaseUrl(session, machineId)}${sessionQuery(session)}`;
}
function sessionQuery(session: SessionLookup): string {
const cwd = sessionCwd(session);
return cwd === undefined || cwd === "" ? "" : `?${new URLSearchParams({ cwd }).toString()}`;
}
function sessionBody(session: SessionLookup, fields: Record<string, unknown> = {}): string {
const cwd = sessionCwd(session);
return JSON.stringify(cwd === undefined || cwd === "" ? fields : { cwd, ...fields });
} }
export const piWebApi = { export const piWebApi = {
@@ -98,26 +121,26 @@ export const workspacesApi = {
export const sessionsApi = { export const sessionsApi = {
sessions: (cwd: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions?cwd=${encodeURIComponent(cwd)}`, arrayOf(parseSessionInfo)), sessions: (cwd: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions?cwd=${encodeURIComponent(cwd)}`, arrayOf(parseSessionInfo)),
startSession: (cwd: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions`, parseSessionInfo, { method: "POST", body: JSON.stringify({ cwd }) }), startSession: (cwd: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions`, parseSessionInfo, { method: "POST", body: JSON.stringify({ cwd }) }),
messages: (session: SessionRef, options?: { limit?: number; before?: number }, machineId = "local") => request(messageUrl(session, options, machineId), parseMessagePage), messages: (session: SessionLookup, options?: { limit?: number; before?: number }, machineId = "local") => request(messageUrl(session, options, machineId), parseMessagePage),
status: (session: SessionRef, machineId = "local") => request(sessionQueryUrl(session, "status", machineId), parseSessionStatus), status: (session: SessionLookup, machineId = "local") => request(sessionQueryUrl(session, "status", machineId), parseSessionStatus),
models: (session: SessionRef, machineId = "local") => request(sessionQueryUrl(session, "models", machineId), parseModelSelectionResponse), models: (session: SessionLookup, machineId = "local") => request(sessionQueryUrl(session, "models", machineId), parseModelSelectionResponse),
setModel: (session: SessionRef, provider: string, modelId: string, machineId = "local") => request(sessionUrl(session, "model", machineId), parseSessionStatus, { method: "POST", body: JSON.stringify({ cwd: session.cwd, provider, modelId }) }), setModel: (session: SessionLookup, provider: string, modelId: string, machineId = "local") => request(sessionUrl(session, "model", machineId), parseSessionStatus, { method: "POST", body: sessionBody(session, { provider, modelId }) }),
cycleModel: (session: SessionRef, direction: "forward" | "backward", machineId = "local") => request(sessionUrl(session, "model/cycle", machineId), parseSessionStatus, { method: "POST", body: JSON.stringify({ cwd: session.cwd, direction }) }), cycleModel: (session: SessionLookup, direction: "forward" | "backward", machineId = "local") => request(sessionUrl(session, "model/cycle", machineId), parseSessionStatus, { method: "POST", body: sessionBody(session, { direction }) }),
thinkingLevels: (session: SessionRef, machineId = "local") => request(sessionQueryUrl(session, "thinking-levels", machineId), parseThinkingLevelsResponse), thinkingLevels: (session: SessionLookup, machineId = "local") => request(sessionQueryUrl(session, "thinking-levels", machineId), parseThinkingLevelsResponse),
setThinkingLevel: (session: SessionRef, level: "off" | "minimal" | "low" | "medium" | "high" | "xhigh", machineId = "local") => request(sessionUrl(session, "thinking-level", machineId), parseSessionStatus, { method: "POST", body: JSON.stringify({ cwd: session.cwd, level }) }), setThinkingLevel: (session: SessionLookup, level: "off" | "minimal" | "low" | "medium" | "high" | "xhigh", machineId = "local") => request(sessionUrl(session, "thinking-level", machineId), parseSessionStatus, { method: "POST", body: sessionBody(session, { level }) }),
cycleThinkingLevel: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "thinking-level/cycle", machineId), parseSessionStatus, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), cycleThinkingLevel: (session: SessionLookup, machineId = "local") => request(sessionUrl(session, "thinking-level/cycle", machineId), parseSessionStatus, { method: "POST", body: sessionBody(session) }),
commands: (session: SessionRef, machineId = "local") => request(sessionQueryUrl(session, "commands", machineId), arrayOf(parseSlashCommand)), commands: (session: SessionLookup, machineId = "local") => request(sessionQueryUrl(session, "commands", machineId), arrayOf(parseSlashCommand)),
prompt: (session: SessionRef, text: string, streamingBehavior?: "steer" | "followUp", machineId = "local") => request(sessionUrl(session, "prompt", machineId), parseAccepted, { method: "POST", body: JSON.stringify(streamingBehavior === undefined ? { cwd: session.cwd, text } : { cwd: session.cwd, text, streamingBehavior }) }), prompt: (session: SessionLookup, text: string, streamingBehavior?: "steer" | "followUp", machineId = "local") => request(sessionUrl(session, "prompt", machineId), parseAccepted, { method: "POST", body: sessionBody(session, streamingBehavior === undefined ? { text } : { text, streamingBehavior }) }),
shell: (session: SessionRef, text: string, machineId = "local") => request(sessionUrl(session, "shell", machineId), parseAccepted, { method: "POST", body: JSON.stringify({ cwd: session.cwd, text }) }), shell: (session: SessionLookup, text: string, machineId = "local") => request(sessionUrl(session, "shell", machineId), parseAccepted, { method: "POST", body: sessionBody(session, { text }) }),
runCommand: (session: SessionRef, text: string, machineId = "local") => request(sessionUrl(session, "commands/run", machineId), parseCommandResult, { method: "POST", body: JSON.stringify({ cwd: session.cwd, text }) }), runCommand: (session: SessionLookup, text: string, machineId = "local") => request(sessionUrl(session, "commands/run", machineId), parseCommandResult, { method: "POST", body: sessionBody(session, { text }) }),
respondToCommand: (session: SessionRef, requestId: string, value: string, machineId = "local") => request(sessionUrl(session, "commands/respond", machineId), parseCommandResult, { method: "POST", body: JSON.stringify({ cwd: session.cwd, requestId, value }) }), respondToCommand: (session: SessionLookup, requestId: string, value: string, machineId = "local") => request(sessionUrl(session, "commands/respond", machineId), parseCommandResult, { method: "POST", body: sessionBody(session, { requestId, value }) }),
abort: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "abort", machineId), parseAborted, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), abort: (session: SessionLookup, machineId = "local") => request(sessionUrl(session, "abort", machineId), parseAborted, { method: "POST", body: sessionBody(session) }),
stop: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "stop", machineId), parseStopped, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), stop: (session: SessionLookup, machineId = "local") => request(sessionUrl(session, "stop", machineId), parseStopped, { method: "POST", body: sessionBody(session) }),
archive: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "archive", machineId), parseArchived, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), archive: (session: SessionLookup, machineId = "local") => request(sessionUrl(session, "archive", machineId), parseArchived, { method: "POST", body: sessionBody(session) }),
archiveWithDescendants: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "archive-tree", machineId), parseArchived, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), archiveWithDescendants: (session: SessionLookup, machineId = "local") => request(sessionUrl(session, "archive-tree", machineId), parseArchived, { method: "POST", body: sessionBody(session) }),
restore: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "restore", machineId), parseRestored, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), restore: (session: SessionLookup, machineId = "local") => request(sessionUrl(session, "restore", machineId), parseRestored, { method: "POST", body: sessionBody(session) }),
deleteArchived: (session: SessionRef, machineId = "local") => request(`${sessionBaseUrl(session, machineId)}?${new URLSearchParams({ cwd: session.cwd }).toString()}`, parseDeleted, { method: "DELETE" }), deleteArchived: (session: SessionLookup, machineId = "local") => request(sessionBaseQueryUrl(session, machineId), parseDeleted, { method: "DELETE" }),
detachParent: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "detach-parent", machineId), parseDetached, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), detachParent: (session: SessionLookup, machineId = "local") => request(sessionUrl(session, "detach-parent", machineId), parseDetached, { method: "POST", body: sessionBody(session) }),
authProviders: (options?: { mode?: "login" | "logout"; authType?: "oauth" | "api_key"; machineId?: string }) => { authProviders: (options?: { mode?: "login" | "logout"; authType?: "oauth" | "api_key"; machineId?: string }) => {
const params = new URLSearchParams(); const params = new URLSearchParams();
if (options?.mode !== undefined) params.set("mode", options.mode); if (options?.mode !== undefined) params.set("mode", options.mode);
+8
View File
@@ -30,6 +30,14 @@ describe("machine-scoped socket urls", () => {
]); ]);
}); });
it("keeps legacy session socket urls usable without cwd", () => {
sessionEvents("s1");
expect(webSocketUrls).toEqual([
"wss://pi.example.test/api/machines/local/sessions/s1/events",
]);
});
it("uses the requested machine scope for terminal sockets", () => { it("uses the requested machine scope for terminal sockets", () => {
terminalSocket("p 1", "w/1", "t?1", { cols: 120, rows: 40 }, "remote-a"); terminalSocket("p 1", "w/1", "t?1", { cols: 120, rows: 40 }, "remote-a");
+7 -3
View File
@@ -1,8 +1,12 @@
import type { SessionRef } from "../../../shared/apiTypes"; import type { SessionRef } from "../../../shared/apiTypes";
export function sessionEvents(session: SessionRef, machineId = "local"): WebSocket { type SessionLookup = SessionRef | string;
const query = new URLSearchParams({ cwd: session.cwd }).toString();
return new WebSocket(`${webSocketBaseUrl()}${machinePrefix(machineId)}/sessions/${encodeURIComponent(session.id)}/events?${query}`); export function sessionEvents(session: SessionLookup, machineId = "local"): WebSocket {
const cwd = typeof session === "string" ? undefined : session.cwd;
const query = cwd === undefined || cwd === "" ? "" : `?${new URLSearchParams({ cwd }).toString()}`;
const sessionId = typeof session === "string" ? session : session.id;
return new WebSocket(`${webSocketBaseUrl()}${machinePrefix(machineId)}/sessions/${encodeURIComponent(sessionId)}/events${query}`);
} }
export function globalSessionEvents(machineId = "local"): WebSocket { export function globalSessionEvents(machineId = "local"): WebSocket {
+16 -3
View File
@@ -1,5 +1,15 @@
import type { SessionRef } from "../../../shared/apiTypes"; import type { SessionRef } from "../../../shared/apiTypes";
type SessionLookup = SessionRef | string;
function sessionId(session: SessionLookup): string {
return typeof session === "string" ? session : session.id;
}
function sessionCwd(session: SessionLookup): string | undefined {
return typeof session === "string" ? undefined : session.cwd;
}
export function gitDiffUrl(projectId: string, workspaceId: string, options?: { path?: string; staged?: boolean }): string { export function gitDiffUrl(projectId: string, workspaceId: string, options?: { path?: string; staged?: boolean }): string {
const params = new URLSearchParams(); const params = new URLSearchParams();
if (options?.path !== undefined) params.set("path", options.path); if (options?.path !== undefined) params.set("path", options.path);
@@ -16,11 +26,14 @@ export function machineGitDiffUrl(machineId: string, projectId: string, workspac
return `/api/machines/${encodeURIComponent(machineId)}/projects/${encodeURIComponent(projectId)}/workspaces/${encodeURIComponent(workspaceId)}/git/diff${query ? `?${query}` : ""}`; return `/api/machines/${encodeURIComponent(machineId)}/projects/${encodeURIComponent(projectId)}/workspaces/${encodeURIComponent(workspaceId)}/git/diff${query ? `?${query}` : ""}`;
} }
export function messageUrl(session: SessionRef, options?: { limit?: number; before?: number }, machineId = "local"): string { export function messageUrl(session: SessionLookup, options?: { limit?: number; before?: number }, machineId = "local"): string {
const params = new URLSearchParams({ cwd: session.cwd }); const params = new URLSearchParams();
const cwd = sessionCwd(session);
if (cwd !== undefined && cwd !== "") params.set("cwd", cwd);
if (options?.limit !== undefined) params.set("limit", String(options.limit)); if (options?.limit !== undefined) params.set("limit", String(options.limit));
if (options?.before !== undefined) params.set("before", String(options.before)); if (options?.before !== undefined) params.set("before", String(options.before));
return `/api/machines/${encodeURIComponent(machineId)}/sessions/${encodeURIComponent(session.id)}/messages?${params.toString()}`; const query = params.toString();
return `/api/machines/${encodeURIComponent(machineId)}/sessions/${encodeURIComponent(sessionId(session))}/messages${query === "" ? "" : `?${query}`}`;
} }
export function workspaceImagePreviewUrl(projectId: string, workspaceId: string, path: string, options?: { modifiedAt?: string; machineId?: string }): string { export function workspaceImagePreviewUrl(projectId: string, workspaceId: string, path: string, options?: { modifiedAt?: string; machineId?: string }): string {
@@ -181,10 +181,10 @@ describe("SessionController", () => {
...defaultApi, ...defaultApi,
startSession: () => Promise.resolve(replacementSession), startSession: () => Promise.resolve(replacementSession),
messages: (session) => { messages: (session) => {
if (session.id === oldSession.id) return Promise.reject(new Error("Session not found")); if (sessionLookupId(session) === oldSession.id) return Promise.reject(new Error("Session not found"));
return Promise.resolve(emptyPage); return Promise.resolve(emptyPage);
}, },
status: (session) => Promise.resolve(status(session.id)), status: (session) => Promise.resolve(status(sessionLookupId(session))),
}; };
const controller = new SessionController( const controller = new SessionController(
() => state, () => state,
@@ -221,7 +221,7 @@ describe("SessionController", () => {
...defaultApi, ...defaultApi,
respondToCommand: () => Promise.resolve({ type: "done", message: "Session forked", session: replacementSession, promptDraft: "fork me" }), respondToCommand: () => Promise.resolve({ type: "done", message: "Session forked", session: replacementSession, promptDraft: "fork me" }),
messages: () => Promise.resolve(emptyPage), messages: () => Promise.resolve(emptyPage),
status: (session) => Promise.resolve(status(session.id)), status: (session) => Promise.resolve(status(sessionLookupId(session))),
}; };
const controller = new SessionController( const controller = new SessionController(
() => state, () => state,
@@ -244,7 +244,7 @@ describe("SessionController", () => {
...defaultApi, ...defaultApi,
archive: () => Promise.resolve({ archived: true }), archive: () => Promise.resolve({ archived: true }),
messages: () => Promise.resolve(emptyPage), messages: () => Promise.resolve(emptyPage),
status: (session) => Promise.resolve(status(session.id)), status: (session) => Promise.resolve(status(sessionLookupId(session))),
}; };
const controller = new SessionController( const controller = new SessionController(
() => state, () => state,
@@ -273,7 +273,7 @@ describe("SessionController", () => {
...defaultApi, ...defaultApi,
archiveWithDescendants: () => Promise.resolve({ archived: true, sessionIds: [oldSession.id, childSession.id], archivedCount: 2, skippedAlreadyArchivedCount: 0 }), archiveWithDescendants: () => Promise.resolve({ archived: true, sessionIds: [oldSession.id, childSession.id], archivedCount: 2, skippedAlreadyArchivedCount: 0 }),
messages: () => Promise.resolve(emptyPage), messages: () => Promise.resolve(emptyPage),
status: (session) => Promise.resolve(status(session.id)), status: (session) => Promise.resolve(status(sessionLookupId(session))),
}; };
const controller = new SessionController( const controller = new SessionController(
() => state, () => state,
@@ -299,11 +299,11 @@ describe("SessionController", () => {
const api: typeof defaultApi = { const api: typeof defaultApi = {
...defaultApi, ...defaultApi,
archive: (session) => { archive: (session) => {
archivedIds.push(session.id); archivedIds.push(sessionLookupId(session));
return Promise.resolve({ archived: true }); return Promise.resolve({ archived: true });
}, },
messages: () => Promise.resolve(emptyPage), messages: () => Promise.resolve(emptyPage),
status: (session) => Promise.resolve(status(session.id)), status: (session) => Promise.resolve(status(sessionLookupId(session))),
}; };
const controller = new SessionController( const controller = new SessionController(
() => state, () => state,
@@ -336,11 +336,11 @@ describe("SessionController", () => {
const api: typeof defaultApi = { const api: typeof defaultApi = {
...defaultApi, ...defaultApi,
deleteArchived: (session) => { deleteArchived: (session) => {
deletedIds.push(session.id); deletedIds.push(sessionLookupId(session));
return Promise.resolve({ deleted: true }); return Promise.resolve({ deleted: true });
}, },
messages: () => Promise.resolve(emptyPage), messages: () => Promise.resolve(emptyPage),
status: (session) => Promise.resolve(status(session.id)), status: (session) => Promise.resolve(status(sessionLookupId(session))),
}; };
const controller = new SessionController( const controller = new SessionController(
() => state, () => state,
@@ -364,7 +364,7 @@ describe("SessionController", () => {
const api: typeof defaultApi = { const api: typeof defaultApi = {
...defaultApi, ...defaultApi,
deleteArchived: (session) => { deleteArchived: (session) => {
deletedIds.push(session.id); deletedIds.push(sessionLookupId(session));
return Promise.resolve({ deleted: true }); return Promise.resolve({ deleted: true });
}, },
}; };
@@ -413,3 +413,7 @@ describe("SessionController", () => {
function sessionKey(sessionId: string): string { function sessionKey(sessionId: string): string {
return machineSessionKey("local", sessionId); return machineSessionKey("local", sessionId);
} }
function sessionLookupId(session: string | SessionRef): string {
return typeof session === "string" ? session : session.id;
}
@@ -60,6 +60,16 @@ describe("SessionDirResolver", () => {
}); });
describe("Pi session manager gateway", () => { describe("Pi session manager gateway", () => {
it("lists legacy id-only sessions from the default Pi session store", async () => {
const otherCwd = join(tempDir, "other-workspace");
await writeSessionFile(defaultPiSessionDir(cwd, agentDir), "session-a", cwd);
await writeSessionFile(defaultPiSessionDir(otherCwd, agentDir), "session-b", otherCwd);
const gateway = createPiSessionManagerGateway({ agentDir, env: {} });
if (gateway.listAll === undefined) throw new Error("Expected legacy listing support");
await expect(gateway.listAll()).resolves.toEqual(expect.arrayContaining([expect.objectContaining({ id: "session-a", cwd }), expect.objectContaining({ id: "session-b", cwd: otherCwd })]));
});
it("lists only sessions for the requested cwd when a custom Pi sessionDir is shared", async () => { it("lists only sessions for the requested cwd when a custom Pi sessionDir is shared", async () => {
const sharedSessionDir = join(tempDir, "shared-sessions"); const sharedSessionDir = join(tempDir, "shared-sessions");
const otherCwd = join(tempDir, "other-workspace"); const otherCwd = join(tempDir, "other-workspace");
+23 -1
View File
@@ -1,3 +1,5 @@
import type { Dirent } from "node:fs";
import { readdir } from "node:fs/promises";
import { homedir } from "node:os"; import { homedir } from "node:os";
import { dirname, isAbsolute, join, resolve } from "node:path"; import { dirname, isAbsolute, join, resolve } from "node:path";
import { getAgentDir, SessionManager, SettingsManager } from "@earendil-works/pi-coding-agent"; import { getAgentDir, SessionManager, SettingsManager } from "@earendil-works/pi-coding-agent";
@@ -27,6 +29,10 @@ export class SessionDirResolver {
this.env = options.env ?? process.env; this.env = options.env ?? process.env;
} }
defaultSessionsRoot(): string {
return defaultPiSessionsRoot(this.agentDir);
}
resolve(cwd: string): SessionDirResolution { resolve(cwd: string): SessionDirResolution {
const envSessionDir = this.env[PI_SESSION_DIR_ENV]; const envSessionDir = this.env[PI_SESSION_DIR_ENV];
if (envSessionDir !== undefined && envSessionDir !== "") { if (envSessionDir !== undefined && envSessionDir !== "") {
@@ -61,6 +67,10 @@ class SettingsAwarePiSessionManagerGateway implements PiSessionManagerGateway {
return SessionManager.create(cwd, resolution.sessionDir); return SessionManager.create(cwd, resolution.sessionDir);
} }
listAll(): Promise<PiSessionListEntry[]> {
return listSessionsInDefaultPiStore(this.resolver.defaultSessionsRoot());
}
open(path: string): PiSessionManager { open(path: string): PiSessionManager {
return SessionManager.open(path, dirname(path)); return SessionManager.open(path, dirname(path));
} }
@@ -70,6 +80,19 @@ export async function listSessionsInDir(sessionDir: string): Promise<PiSessionLi
return SessionManager.list("", sessionDir); return SessionManager.list("", sessionDir);
} }
export async function listSessionsInDefaultPiStore(storeRoot = defaultPiSessionsRoot()): Promise<PiSessionListEntry[]> {
let entries: Dirent[];
try {
entries = await readdir(storeRoot, { withFileTypes: true });
} catch {
return [];
}
const sessionDirs = entries.filter((entry) => entry.isDirectory()).map((entry) => join(storeRoot, entry.name));
const sessions = (await Promise.all(sessionDirs.map((dir) => listSessionsInDir(dir)))).flat();
return sessions.sort((a, b) => b.modified.getTime() - a.modified.getTime());
}
export function filterSessionsForCwd(sessions: readonly PiSessionListEntry[], cwd: string): PiSessionListEntry[] { export function filterSessionsForCwd(sessions: readonly PiSessionListEntry[], cwd: string): PiSessionListEntry[] {
return sessions.filter((session) => session.cwd === cwd); return sessions.filter((session) => session.cwd === cwd);
} }
@@ -97,4 +120,3 @@ function expandTildePath(path: string): string {
if (path.startsWith("~/")) return join(homedir(), path.slice(2)); if (path.startsWith("~/")) return join(homedir(), path.slice(2));
return path; return path;
} }
@@ -164,6 +164,27 @@ describe("PiSessionService", () => {
expect(fake.calls.dispose).toBe(1); expect(fake.calls.dispose).toBe(1);
}); });
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("binds extensions again when the SDK runtime replaces the active session", async () => { it("binds extensions again when the SDK runtime replaces the active session", async () => {
const hub = new CapturingSessionEventHub(); const hub = new CapturingSessionEventHub();
const fake = fakeRuntime("session-1"); const fake = fakeRuntime("session-1");
+9 -2
View File
@@ -103,6 +103,12 @@ export interface PiSessionManager {
export interface PiSessionManagerGateway { export interface PiSessionManagerGateway {
list(cwd: string): Promise<PiSessionListEntry[]>; list(cwd: string): Promise<PiSessionListEntry[]>;
create(cwd: string): PiSessionManager; create(cwd: string): PiSessionManager;
/**
* Legacy id-only lookup surface for older clients. This intentionally searches
* only Pi's default session store, because custom session directories require
* a cwd-scoped lookup.
*/
listAll?(): Promise<PiSessionListEntry[]>;
open(path: string): PiSessionManager; open(path: string): PiSessionManager;
} }
@@ -671,8 +677,9 @@ export class PiSessionService {
const archived = await this.getArchived(ref); const archived = await this.getArchived(ref);
if (archived?.archivePath !== undefined) return this.create(this.sessionManager.open(archived.archivePath), archived.cwd); if (archived?.archivePath !== undefined) return this.create(this.sessionManager.open(archived.archivePath), archived.cwd);
if (!isPiSessionRef(ref)) throw new Error("Session not found"); const match = isPiSessionRef(ref)
const match = (await this.sessionManager.list(ref.cwd)).find((s) => s.id === ref.id || s.id.startsWith(ref.id)); ? (await this.sessionManager.list(ref.cwd)).find((s) => s.id === ref.id || s.id.startsWith(ref.id))
: (await this.sessionManager.listAll?.() ?? []).find((s) => s.id === ref || s.id.startsWith(ref));
if (!match) throw new Error("Session not found"); if (!match) throw new Error("Session not found");
return this.create(this.sessionManager.open(match.path), match.cwd); return this.create(this.sessionManager.open(match.path), match.cwd);
} }
+73 -2
View File
@@ -2,7 +2,7 @@ import Fastify, { type FastifyInstance } from "fastify";
import fastifyWebsocket from "@fastify/websocket"; import fastifyWebsocket from "@fastify/websocket";
import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { SessionEventHub } from "../realtime/sessionEventHub.js"; import { SessionEventHub } from "../realtime/sessionEventHub.js";
import { PiSessionService, type PiSessionManagerGateway } from "./piSessionService.js"; import { PiSessionService, type PiSessionManagerGateway, type PiSessionRef } from "./piSessionService.js";
import { registerSessionRoutes } from "./sessionRoutes.js"; import { registerSessionRoutes } from "./sessionRoutes.js";
let app: FastifyInstance; let app: FastifyInstance;
@@ -25,14 +25,81 @@ afterEach(async () => {
describe("session routes", () => { describe("session routes", () => {
it("rejects prompt payloads that omit text without opening a session", async () => { it("rejects prompt payloads that omit text without opening a session", async () => {
const response = await app.inject({ method: "POST", url: "/sessions/session-1/prompt", payload: { cwd: "/repo", body: "Build the thing" } }); const response = await app.inject({ method: "POST", url: "/sessions/session-1/prompt", payload: { body: "Build the thing" } });
expect(response.statusCode).toBe(400); expect(response.statusCode).toBe(400);
expect(response.json()).toEqual({ error: "Prompt text is required" }); expect(response.json()).toEqual({ error: "Prompt text is required" });
expect(sessionManager.calls).toEqual({ create: 0, list: 0, listAll: 0, open: 0 }); expect(sessionManager.calls).toEqual({ create: 0, list: 0, listAll: 0, open: 0 });
}); });
it("keeps legacy per-session routes usable without cwd", async () => {
const routeApp = Fastify({ logger: false });
await routeApp.register(fastifyWebsocket);
const eventHub = new SessionEventHub();
const routeService = new CapturingRouteSessionService(eventHub);
registerSessionRoutes(routeApp, routeService, eventHub);
try {
const statusResponse = await routeApp.inject({ method: "GET", url: "/sessions/session-1/status" });
const promptResponse = await routeApp.inject({ method: "POST", url: "/sessions/session-1/prompt", payload: { text: "hello" } });
expect(statusResponse.statusCode).toBe(200);
expect(promptResponse.statusCode).toBe(200);
expect(routeService.calls).toEqual(["session-1", { lookup: "session-1", text: "hello" }]);
} finally {
await routeService.dispose();
await routeApp.close();
}
});
it("passes cwd when per-session routes include workspace context", async () => {
const routeApp = Fastify({ logger: false });
await routeApp.register(fastifyWebsocket);
const eventHub = new SessionEventHub();
const routeService = new CapturingRouteSessionService(eventHub);
registerSessionRoutes(routeApp, routeService, eventHub);
try {
const statusResponse = await routeApp.inject({ method: "GET", url: `/sessions/session-1/status?cwd=${encodeURIComponent("/repo")}` });
const promptResponse = await routeApp.inject({ method: "POST", url: "/sessions/session-1/prompt", payload: { cwd: "/repo", text: "hello" } });
expect(statusResponse.statusCode).toBe(200);
expect(promptResponse.statusCode).toBe(200);
expect(routeService.calls).toEqual([{ id: "session-1", cwd: "/repo" }, { lookup: { id: "session-1", cwd: "/repo" }, text: "hello" }]);
} finally {
await routeService.dispose();
await routeApp.close();
}
});
}); });
class CapturingRouteSessionService extends PiSessionService {
readonly calls: unknown[] = [];
constructor(eventHub: SessionEventHub) {
super(eventHub, { sessionManager: new RejectingSessionManager(), heartbeatIntervalMs: 60_000 });
}
override status(lookup: string | PiSessionRef) {
this.calls.push(lookup);
return Promise.resolve({
sessionId: sessionIdFromLookup(lookup),
isStreaming: false,
isCompacting: false,
isBashRunning: false,
pendingMessageCount: 0,
queuedMessages: [],
tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
cost: 0,
});
}
override prompt(lookup: string | PiSessionRef, text: unknown): Promise<void> {
this.calls.push({ lookup, text });
return Promise.resolve();
}
}
class RejectingSessionManager implements PiSessionManagerGateway { class RejectingSessionManager implements PiSessionManagerGateway {
readonly calls = { create: 0, list: 0, listAll: 0, open: 0 }; readonly calls = { create: 0, list: 0, listAll: 0, open: 0 };
@@ -56,3 +123,7 @@ class RejectingSessionManager implements PiSessionManagerGateway {
throw new Error("Session manager should not open sessions for invalid prompt payloads"); throw new Error("Session manager should not open sessions for invalid prompt payloads");
} }
} }
function sessionIdFromLookup(lookup: string | PiSessionRef): string {
return typeof lookup === "string" ? lookup : lookup.id;
}
+94 -80
View File
@@ -2,6 +2,8 @@ import type { FastifyInstance } from "fastify";
import type { SessionEventHub } from "../realtime/sessionEventHub.js"; import type { SessionEventHub } from "../realtime/sessionEventHub.js";
import type { PiSessionRef, PiSessionService } from "./piSessionService.js"; import type { PiSessionRef, PiSessionService } from "./piSessionService.js";
type SessionLookup = string | PiSessionRef;
interface SessionQuery { interface SessionQuery {
cwd?: string; cwd?: string;
} }
@@ -11,8 +13,6 @@ interface MessageQuery extends SessionQuery {
limit?: string; limit?: string;
} }
class SessionRouteValidationError extends Error {}
interface PromptRequestBody { interface PromptRequestBody {
cwd?: unknown; cwd?: unknown;
text?: unknown; text?: unknown;
@@ -25,9 +25,10 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS
return sessions.list(request.query.cwd); return sessions.list(request.query.cwd);
}); });
app.post<{ Body: { cwd: string } }>(`${prefix}/sessions`, async (request, reply) => { app.post<{ Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions`, async (request, reply) => {
try { try {
return await sessions.start(request.body.cwd); const body = requireRecord(request.body);
return await sessions.start(requireString(body, "cwd"));
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(400).send({ error: errorMessage(error) });
} }
@@ -36,189 +37,185 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS
app.get<{ Params: { sessionId: string }; Querystring: MessageQuery }>(`${prefix}/sessions/:sessionId/messages`, async (request, reply) => { app.get<{ Params: { sessionId: string }; Querystring: MessageQuery }>(`${prefix}/sessions/:sessionId/messages`, async (request, reply) => {
try { try {
const page = { ...optionalField("before", optionalNumber(request.query.before)), ...optionalField("limit", optionalNumber(request.query.limit)) }; const page = { ...optionalField("before", optionalNumber(request.query.before)), ...optionalField("limit", optionalNumber(request.query.limit)) };
return await sessions.messages(sessionRefFromQuery(request.params.sessionId, request.query), page); return await sessions.messages(sessionLookupFromQuery(request.params.sessionId, request.query), page);
} catch (error) { } catch (error) {
return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); return reply.code(404).send({ error: errorMessage(error) });
} }
}); });
app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/status`, async (request, reply) => { app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/status`, async (request, reply) => {
try { try {
return await sessions.status(sessionRefFromQuery(request.params.sessionId, request.query)); return await sessions.status(sessionLookupFromQuery(request.params.sessionId, request.query));
} catch (error) { } catch (error) {
return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); return reply.code(404).send({ error: errorMessage(error) });
} }
}); });
app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/models`, async (request, reply) => { app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/models`, async (request, reply) => {
try { try {
return { models: await sessions.availableModels(sessionRefFromQuery(request.params.sessionId, request.query)) }; return { models: await sessions.availableModels(sessionLookupFromQuery(request.params.sessionId, request.query)) };
} catch (error) { } catch (error) {
return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); return reply.code(404).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; provider?: unknown; modelId?: unknown } }>(`${prefix}/sessions/:sessionId/model`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; provider?: unknown; modelId?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/model`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
return await sessions.setModel(sessionRefFromBody(request.params.sessionId, body), requireString(body, "provider"), requireString(body, "modelId")); return await sessions.setModel(sessionLookupFromBody(request.params.sessionId, body), requireString(body, "provider"), requireString(body, "modelId"));
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; direction?: "forward" | "backward" } }>(`${prefix}/sessions/:sessionId/model/cycle`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; direction?: "forward" | "backward" } | undefined }>(`${prefix}/sessions/:sessionId/model/cycle`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
const direction = body["direction"]; const direction = body["direction"];
if (direction !== undefined && direction !== "forward" && direction !== "backward") throw new Error("direction must be forward or backward"); if (direction !== undefined && direction !== "forward" && direction !== "backward") throw new Error("direction must be forward or backward");
return await sessions.cycleModel(sessionRefFromBody(request.params.sessionId, body), direction ?? "forward"); return await sessions.cycleModel(sessionLookupFromBody(request.params.sessionId, body), direction ?? "forward");
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/thinking-levels`, async (request, reply) => { app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/thinking-levels`, async (request, reply) => {
try { try {
return { levels: await sessions.availableThinkingLevels(sessionRefFromQuery(request.params.sessionId, request.query)) }; return { levels: await sessions.availableThinkingLevels(sessionLookupFromQuery(request.params.sessionId, request.query)) };
} catch (error) { } catch (error) {
return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); return reply.code(404).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; level?: unknown } }>(`${prefix}/sessions/:sessionId/thinking-level`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; level?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/thinking-level`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
return await sessions.setThinkingLevel(sessionRefFromBody(request.params.sessionId, body), requireThinkingLevel(body["level"])); return await sessions.setThinkingLevel(sessionLookupFromBody(request.params.sessionId, body), requireThinkingLevel(body["level"]));
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/thinking-level/cycle`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/thinking-level/cycle`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
return await sessions.cycleThinkingLevel(sessionRefFromBody(request.params.sessionId, body)); return await sessions.cycleThinkingLevel(sessionLookupFromBody(request.params.sessionId, body));
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/commands`, async (request, reply) => { app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/commands`, async (request, reply) => {
try { try {
return await sessions.commands(sessionRefFromQuery(request.params.sessionId, request.query)); return await sessions.commands(sessionLookupFromQuery(request.params.sessionId, request.query));
} catch (error) { } catch (error) {
return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); return reply.code(404).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: PromptRequestBody | undefined }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: PromptRequestBody | undefined }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
await sessions.prompt(sessionRefFromBody(request.params.sessionId, body), body["text"], body["streamingBehavior"]); await sessions.prompt(sessionLookupFromBody(request.params.sessionId, body), body["text"], body["streamingBehavior"]);
return { accepted: true }; return { accepted: true };
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown } }>(`${prefix}/sessions/:sessionId/shell`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/shell`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
await sessions.shell(sessionRefFromBody(request.params.sessionId, body), requireString(body, "text")); await sessions.shell(sessionLookupFromBody(request.params.sessionId, body), requireString(body, "text"));
return { accepted: true }; return { accepted: true };
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown } }>(`${prefix}/sessions/:sessionId/commands/run`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/commands/run`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
return await sessions.runCommand(sessionRefFromBody(request.params.sessionId, body), requireString(body, "text")); return await sessions.runCommand(sessionLookupFromBody(request.params.sessionId, body), requireString(body, "text"));
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; requestId?: unknown; value?: unknown } }>(`${prefix}/sessions/:sessionId/commands/respond`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; requestId?: unknown; value?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/commands/respond`, async (request, reply) => {
try { try {
const body = requireRecord(request.body); const body = optionalRecord(request.body);
return await sessions.respondToCommand(sessionRefFromBody(request.params.sessionId, body), requireString(body, "requestId"), requireString(body, "value")); return await sessions.respondToCommand(sessionLookupFromBody(request.params.sessionId, body), requireString(body, "requestId"), requireString(body, "value"));
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/abort`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/abort`, async (request, reply) => {
try { try {
await sessions.abort(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); await sessions.abort(sessionLookupFromBody(request.params.sessionId, optionalRecord(request.body)));
return { aborted: true }; return { aborted: true };
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/stop`, (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/stop`, (request, reply) => {
try { try {
sessions.stop(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); sessions.stop(sessionLookupFromBody(request.params.sessionId, optionalRecord(request.body)));
return { stopped: true }; return { stopped: true };
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/archive`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/archive`, async (request, reply) => {
try { try {
await sessions.archive(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); await sessions.archive(sessionLookupFromBody(request.params.sessionId, optionalRecord(request.body)));
return { archived: true }; return { archived: true };
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/archive-tree`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/archive-tree`, async (request, reply) => {
try { try {
return await sessions.archiveTree(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); return await sessions.archiveTree(sessionLookupFromBody(request.params.sessionId, optionalRecord(request.body)));
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/restore`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/restore`, async (request, reply) => {
try { try {
await sessions.restore(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); await sessions.restore(sessionLookupFromBody(request.params.sessionId, optionalRecord(request.body)));
return { restored: true }; return { restored: true };
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.delete<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId`, async (request, reply) => { app.delete<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId`, async (request, reply) => {
try { try {
await sessions.deleteArchived(sessionRefFromQuery(request.params.sessionId, request.query)); await sessions.deleteArchived(sessionLookupFromQuery(request.params.sessionId, request.query));
return { deleted: true }; return { deleted: true };
} catch (error) { } catch (error) {
return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/detach-parent`, async (request, reply) => { app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/detach-parent`, async (request, reply) => {
try { try {
await sessions.detachParent(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); await sessions.detachParent(sessionLookupFromBody(request.params.sessionId, optionalRecord(request.body)));
return { detached: true }; return { detached: true };
} catch (error) { } catch (error) {
return reply.code(400).send({ error: errorMessage(error) }); return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) });
} }
}); });
app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/events`, { websocket: true }, (socket, request) => { app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/events`, { websocket: true }, (socket, request) => {
try { const lookup = sessionLookupFromQuery(request.params.sessionId, request.query);
const ref = sessionRefFromQuery(request.params.sessionId, request.query); eventHub.add(sessionIdFromLookup(lookup), socket);
eventHub.add(ref.id, socket);
} catch {
socket.close();
}
}); });
app.get(`${prefix}/sessions/events`, { websocket: true }, (socket) => { app.get(`${prefix}/sessions/events`, { websocket: true }, (socket) => {
@@ -230,16 +227,28 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS
}); });
} }
function sessionRefFromQuery(id: string, query: SessionQuery): PiSessionRef { function sessionLookupFromQuery(id: string, query: SessionQuery): SessionLookup {
const cwd = query.cwd; return sessionLookupFromCwd(id, query.cwd);
if (cwd === undefined || cwd === "") throw new SessionRouteValidationError("cwd query parameter is required"); }
function sessionLookupFromBody(id: string, body: Record<string, unknown>): SessionLookup {
const cwd = body["cwd"];
if (cwd === undefined || cwd === "") return id;
if (typeof cwd !== "string") throw new Error("cwd field must be a string");
return { id, cwd }; return { id, cwd };
} }
function sessionRefFromBody(id: string, body: Record<string, unknown>): PiSessionRef { function sessionLookupFromCwd(id: string, cwd: string | undefined): SessionLookup {
const cwd = body["cwd"]; return cwd === undefined || cwd === "" ? id : { id, cwd };
if (typeof cwd !== "string" || cwd === "") throw new Error("cwd field is required"); }
return { id, cwd };
function sessionIdFromLookup(lookup: SessionLookup): string {
return typeof lookup === "string" ? lookup : lookup.id;
}
function optionalRecord(value: unknown): Record<string, unknown> {
if (value === undefined || value === null) return {};
return requireRecord(value);
} }
function requireRecord(value: unknown): Record<string, unknown> { function requireRecord(value: unknown): Record<string, unknown> {
@@ -272,8 +281,13 @@ function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error); return error instanceof Error ? error.message : String(error);
} }
function readErrorStatus(error: unknown): 400 | 404 { function mutationErrorStatus(error: unknown): 400 | 404 {
return error instanceof SessionRouteValidationError ? 400 : 404; return isSessionNotFoundError(error) ? 404 : 400;
}
function isSessionNotFoundError(error: unknown): boolean {
const message = errorMessage(error);
return message === "Session not found" || message === "Archived session not found";
} }
function isRecord(value: unknown): value is Record<string, unknown> { function isRecord(value: unknown): value is Record<string, unknown> {