From 06052ea5ec492dec18ee26411304838e7f6e8642 Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Wed, 10 Jun 2026 20:26:51 +0200 Subject: [PATCH] fix: respect Pi session directories by cwd --- .changeset/respect-pi-session-dir.md | 5 + README.md | 2 + docs/install.html | 2 + src/client/src/api.ts | 2 +- src/client/src/api/clients.ts | 49 +++-- .../src/api/federatedRouteContract.test.ts | 41 ++-- src/client/src/api/sockets.test.ts | 4 +- src/client/src/api/sockets.ts | 7 +- src/client/src/api/urls.ts | 9 +- src/client/src/components/PromptEditor.ts | 4 +- src/client/src/controllers/authController.ts | 10 +- .../src/controllers/sessionController.test.ts | 18 +- .../src/controllers/sessionController.ts | 50 +++-- src/client/src/sessionSocket.ts | 14 +- src/server/sessiond/sessionProxyRoutes.ts | 2 +- .../sessions/piSessionManagerGateway.test.ts | 85 +++++++ .../sessions/piSessionManagerGateway.ts | 100 +++++++++ src/server/sessions/piSessionService.test.ts | 51 +++-- src/server/sessions/piSessionService.ts | 164 +++++++++----- src/server/sessions/sessionRoutes.ts | 208 ++++++++++++------ src/server/types.ts | 1 + src/shared/apiTypes.ts | 7 +- 22 files changed, 583 insertions(+), 252 deletions(-) create mode 100644 .changeset/respect-pi-session-dir.md create mode 100644 src/server/sessions/piSessionManagerGateway.test.ts create mode 100644 src/server/sessions/piSessionManagerGateway.ts diff --git a/.changeset/respect-pi-session-dir.md b/.changeset/respect-pi-session-dir.md new file mode 100644 index 0000000..e944ccb --- /dev/null +++ b/.changeset/respect-pi-session-dir.md @@ -0,0 +1,5 @@ +--- +"@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. diff --git a/README.md b/README.md index 1a91c11..4dc0a72 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,8 @@ Environment variables: - `PI_WEB_SESSIOND_URL` — daemon URL used by the web process when connecting over TCP, for example `http://127.0.0.1:3001`. If you set `PI_WEB_SESSIOND_PORT`, set this for the web process too. - `PI_WEB_PROJECTS_FILE` — optional override for the projects storage JSON file. Defaults to `$PI_WEB_DATA_DIR/projects.json`. - `PI_WEB_MACHINES_FILE` — optional override for the remote machine registry JSON file. Defaults to `$PI_WEB_DATA_DIR/machines.json`. +- `PI_CODING_AGENT_SESSION_DIR` — Pi session storage directory. PI WEB follows the same session-location priority as Pi for web sessions: this environment variable, then `sessionDir` in Pi settings for the selected workspace, then Pi's default session directory. +- `PI_CODING_AGENT_DIR` — Pi agent config directory. PI WEB uses this for Pi auth, settings, resources, and default session storage, matching Pi's own configuration layout. ## Development services diff --git a/docs/install.html b/docs/install.html index c469876..ac60a54 100644 --- a/docs/install.html +++ b/docs/install.html @@ -241,6 +241,8 @@
  • PI_WEB_HOST: web server bind host. Overrides the config file. Use 127.0.0.1 for local/tunnel-only access, or a specific VPN/private-network IP for trusted remote access.
  • PI_WEB_DATA_DIR: data directory, default ~/.pi-web.
  • PI_WEB_SESSIOND_SOCKET: Unix socket path for daemon communication.
    • +
  • PI_CODING_AGENT_SESSION_DIR: Pi session storage directory. PI WEB follows Pi's priority for sessions: this environment variable, then sessionDir in Pi settings for the selected workspace, then Pi's default session directory.
    • +
  • PI_CODING_AGENT_DIR: Pi agent config directory for auth, settings, resources, and default session storage.
    • diff --git a/src/client/src/api.ts b/src/client/src/api.ts index 547421a..d7b5a48 100644 --- a/src/client/src/api.ts +++ b/src/client/src/api.ts @@ -1,3 +1,3 @@ export { activityApi, api, configApi, filesApi, gitApi, machinesApi, piWebApi, pluginsApi, projectsApi, sessionsApi, terminalsApi, workspacesApi } from "./api/clients"; export { globalSessionEvents, realtimeEvents, sessionEvents, terminalSocket } from "./api/sockets"; -export type { ArchiveSessionsResponse, AuthProviderOption, AuthProviderStatus, AuthProvidersResponse, AuthStatusSource, AuthType, CommandOption, CommandResult, FileContentMediaType, FileContentResponse, FileSuggestion, FileTreeEntry, FileTreeResponse, GitDiffResponse, GitFileState, GitStatusFile, GitStatusResponse, Machine, MachineHealth, MachineKind, MachineStatus, MessagePage, ModelSelectionResponse, OAuthFlowState, PiWebComponentStatus, PiWebConfigEnvOverrides, PiWebConfigResponse, PiWebConfigValues, PiWebInstallationInfo, PiWebPluginConfig, PiWebPluginConfigMap, PiWebPluginInfo, PiWebPluginsResponse, PiWebPluginScope, PiWebPluginSettings, PiWebReleaseStatus, PiWebShortcutConfig, PiWebStatusMessage, PiWebStatusResponse, Project, QueuedSessionMessage, RealtimeEvent, RunTerminalCommandInput, SessionActivity, SessionInfo, SessionModel, SessionStatus, SlashCommand, SessionUiEvent, TerminalCommandRun, TerminalCommandRunFilter, TerminalCommandRunHandle, TerminalCommandRunStatus, TerminalInfo, TerminalUiEvent, ThinkingLevel, ThinkingLevelsResponse, Workspace, WorkspaceActivity, WorkspaceActivityResponse, WorkspaceActivityUiEvent } from "../../shared/apiTypes"; +export type { ArchiveSessionsResponse, AuthProviderOption, AuthProviderStatus, AuthProvidersResponse, AuthStatusSource, AuthType, CommandOption, CommandResult, FileContentMediaType, FileContentResponse, FileSuggestion, FileTreeEntry, FileTreeResponse, GitDiffResponse, GitFileState, GitStatusFile, GitStatusResponse, Machine, MachineHealth, MachineKind, MachineStatus, MessagePage, ModelSelectionResponse, OAuthFlowState, PiWebComponentStatus, PiWebConfigEnvOverrides, PiWebConfigResponse, PiWebConfigValues, PiWebInstallationInfo, PiWebPluginConfig, PiWebPluginConfigMap, PiWebPluginInfo, PiWebPluginsResponse, PiWebPluginScope, PiWebPluginSettings, PiWebReleaseStatus, PiWebShortcutConfig, PiWebStatusMessage, PiWebStatusResponse, Project, QueuedSessionMessage, RealtimeEvent, RunTerminalCommandInput, SessionActivity, SessionInfo, SessionRef, SessionModel, SessionStatus, SlashCommand, SessionUiEvent, TerminalCommandRun, TerminalCommandRunFilter, TerminalCommandRunHandle, TerminalCommandRunStatus, TerminalInfo, TerminalUiEvent, ThinkingLevel, ThinkingLevelsResponse, Workspace, WorkspaceActivity, WorkspaceActivityResponse, WorkspaceActivityUiEvent } from "../../shared/apiTypes"; diff --git a/src/client/src/api/clients.ts b/src/client/src/api/clients.ts index aab597e..8b85453 100644 --- a/src/client/src/api/clients.ts +++ b/src/client/src/api/clients.ts @@ -1,4 +1,4 @@ -import type { FileSuggestion, PiWebConfigValues, RunTerminalCommandInput, TerminalCommandRun, TerminalCommandRunFilter } from "../../../shared/apiTypes"; +import type { FileSuggestion, PiWebConfigValues, RunTerminalCommandInput, SessionRef, TerminalCommandRun, TerminalCommandRunFilter } from "../../../shared/apiTypes"; import { request } from "./http"; import { arrayOf, @@ -39,6 +39,15 @@ import { machineGitDiffUrl, messageUrl } from "./urls"; const machinePrefix = (machineId = "local") => `/api/machines/${encodeURIComponent(machineId)}`; +function sessionUrl(session: SessionRef, endpoint: string, machineId = "local"): string { + return `${machinePrefix(machineId)}/sessions/${encodeURIComponent(session.id)}/${endpoint}`; +} + +function sessionQueryUrl(session: SessionRef, endpoint: string, machineId = "local"): string { + const query = new URLSearchParams({ cwd: session.cwd }).toString(); + return `${sessionUrl(session, endpoint, machineId)}?${query}`; +} + export const piWebApi = { piWebStatus: () => request("/api/pi-web/status", parsePiWebStatusResponse), }; @@ -80,25 +89,25 @@ export const workspacesApi = { export const sessionsApi = { 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 }) }), - messages: (sessionId: string, options?: { limit?: number; before?: number }, machineId = "local") => request(messageUrl(sessionId, options, machineId), parseMessagePage), - status: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/status`, parseSessionStatus), - models: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/models`, parseModelSelectionResponse), - setModel: (sessionId: string, provider: string, modelId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/model`, parseSessionStatus, { method: "POST", body: JSON.stringify({ provider, modelId }) }), - cycleModel: (sessionId: string, direction: "forward" | "backward", machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/model/cycle`, parseSessionStatus, { method: "POST", body: JSON.stringify({ direction }) }), - thinkingLevels: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/thinking-levels`, parseThinkingLevelsResponse), - setThinkingLevel: (sessionId: string, level: "off" | "minimal" | "low" | "medium" | "high" | "xhigh", machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/thinking-level`, parseSessionStatus, { method: "POST", body: JSON.stringify({ level }) }), - cycleThinkingLevel: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/thinking-level/cycle`, parseSessionStatus, { method: "POST" }), - commands: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/commands`, arrayOf(parseSlashCommand)), - prompt: (sessionId: string, text: string, streamingBehavior?: "steer" | "followUp", machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/prompt`, parseAccepted, { method: "POST", body: JSON.stringify(streamingBehavior === undefined ? { text } : { text, streamingBehavior }) }), - shell: (sessionId: string, text: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/shell`, parseAccepted, { method: "POST", body: JSON.stringify({ text }) }), - runCommand: (sessionId: string, text: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/commands/run`, parseCommandResult, { method: "POST", body: JSON.stringify({ text }) }), - respondToCommand: (sessionId: string, requestId: string, value: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/commands/respond`, parseCommandResult, { method: "POST", body: JSON.stringify({ requestId, value }) }), - abort: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/abort`, parseAborted, { method: "POST" }), - stop: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/stop`, parseStopped, { method: "POST" }), - archive: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/archive`, parseArchived, { method: "POST" }), - archiveWithDescendants: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/archive-tree`, parseArchived, { method: "POST" }), - restore: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/restore`, parseRestored, { method: "POST" }), - detachParent: (sessionId: string, machineId = "local") => request(`${machinePrefix(machineId)}/sessions/${sessionId}/detach-parent`, parseDetached, { method: "POST" }), + messages: (session: SessionRef, options?: { limit?: number; before?: number }, machineId = "local") => request(messageUrl(session, options, machineId), parseMessagePage), + status: (session: SessionRef, machineId = "local") => request(sessionQueryUrl(session, "status", machineId), parseSessionStatus), + models: (session: SessionRef, 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 }) }), + cycleModel: (session: SessionRef, direction: "forward" | "backward", machineId = "local") => request(sessionUrl(session, "model/cycle", machineId), parseSessionStatus, { method: "POST", body: JSON.stringify({ cwd: session.cwd, direction }) }), + thinkingLevels: (session: SessionRef, 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 }) }), + cycleThinkingLevel: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "thinking-level/cycle", machineId), parseSessionStatus, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), + commands: (session: SessionRef, 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 }) }), + shell: (session: SessionRef, text: string, machineId = "local") => request(sessionUrl(session, "shell", machineId), parseAccepted, { method: "POST", body: JSON.stringify({ cwd: session.cwd, text }) }), + runCommand: (session: SessionRef, text: string, machineId = "local") => request(sessionUrl(session, "commands/run", machineId), parseCommandResult, { method: "POST", body: JSON.stringify({ cwd: session.cwd, 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 }) }), + abort: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "abort", machineId), parseAborted, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), + stop: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "stop", machineId), parseStopped, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), + archive: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "archive", machineId), parseArchived, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), + archiveWithDescendants: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "archive-tree", machineId), parseArchived, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), + restore: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "restore", machineId), parseRestored, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), + detachParent: (session: SessionRef, machineId = "local") => request(sessionUrl(session, "detach-parent", machineId), parseDetached, { method: "POST", body: JSON.stringify({ cwd: session.cwd }) }), authProviders: (options?: { mode?: "login" | "logout"; authType?: "oauth" | "api_key"; machineId?: string }) => { const params = new URLSearchParams(); if (options?.mode !== undefined) params.set("mode", options.mode); diff --git a/src/client/src/api/federatedRouteContract.test.ts b/src/client/src/api/federatedRouteContract.test.ts index 5104513..2e1188b 100644 --- a/src/client/src/api/federatedRouteContract.test.ts +++ b/src/client/src/api/federatedRouteContract.test.ts @@ -15,6 +15,7 @@ const workspace: Workspace = { isGitRepo: true, isGitWorktree: true, }; +const session = { id: "s 1", cwd: workspace.path }; afterEach(() => { vi.unstubAllGlobals(); @@ -40,25 +41,25 @@ describe("federated route contract", () => { ignoreParseFailure(gitApi.gitDiff("p 1", "w 1", { path: "README.md", staged: true }, machineId)), ignoreParseFailure(sessionsApi.sessions("/repo", machineId)), ignoreParseFailure(sessionsApi.startSession("/repo", machineId)), - ignoreParseFailure(sessionsApi.messages("s 1", { limit: 20, before: 10 }, machineId)), - ignoreParseFailure(sessionsApi.status("s 1", machineId)), - ignoreParseFailure(sessionsApi.models("s 1", machineId)), - ignoreParseFailure(sessionsApi.setModel("s 1", "openai", "gpt", machineId)), - ignoreParseFailure(sessionsApi.cycleModel("s 1", "forward", machineId)), - ignoreParseFailure(sessionsApi.thinkingLevels("s 1", machineId)), - ignoreParseFailure(sessionsApi.setThinkingLevel("s 1", "medium", machineId)), - ignoreParseFailure(sessionsApi.cycleThinkingLevel("s 1", machineId)), - ignoreParseFailure(sessionsApi.commands("s 1", machineId)), - ignoreParseFailure(sessionsApi.prompt("s 1", "hello", "followUp", machineId)), - ignoreParseFailure(sessionsApi.shell("s 1", "ls", machineId)), - ignoreParseFailure(sessionsApi.runCommand("s 1", "/help", machineId)), - ignoreParseFailure(sessionsApi.respondToCommand("s 1", "req 1", "yes", machineId)), - ignoreParseFailure(sessionsApi.abort("s 1", machineId)), - ignoreParseFailure(sessionsApi.stop("s 1", machineId)), - ignoreParseFailure(sessionsApi.archive("s 1", machineId)), - ignoreParseFailure(sessionsApi.archiveWithDescendants("s 1", machineId)), - ignoreParseFailure(sessionsApi.restore("s 1", machineId)), - ignoreParseFailure(sessionsApi.detachParent("s 1", machineId)), + ignoreParseFailure(sessionsApi.messages(session, { limit: 20, before: 10 }, machineId)), + ignoreParseFailure(sessionsApi.status(session, machineId)), + ignoreParseFailure(sessionsApi.models(session, machineId)), + ignoreParseFailure(sessionsApi.setModel(session, "openai", "gpt", machineId)), + ignoreParseFailure(sessionsApi.cycleModel(session, "forward", machineId)), + ignoreParseFailure(sessionsApi.thinkingLevels(session, machineId)), + ignoreParseFailure(sessionsApi.setThinkingLevel(session, "medium", machineId)), + ignoreParseFailure(sessionsApi.cycleThinkingLevel(session, machineId)), + ignoreParseFailure(sessionsApi.commands(session, machineId)), + ignoreParseFailure(sessionsApi.prompt(session, "hello", "followUp", machineId)), + ignoreParseFailure(sessionsApi.shell(session, "ls", machineId)), + ignoreParseFailure(sessionsApi.runCommand(session, "/help", machineId)), + ignoreParseFailure(sessionsApi.respondToCommand(session, "req 1", "yes", machineId)), + ignoreParseFailure(sessionsApi.abort(session, machineId)), + ignoreParseFailure(sessionsApi.stop(session, machineId)), + ignoreParseFailure(sessionsApi.archive(session, machineId)), + ignoreParseFailure(sessionsApi.archiveWithDescendants(session, machineId)), + ignoreParseFailure(sessionsApi.restore(session, machineId)), + ignoreParseFailure(sessionsApi.detachParent(session, machineId)), ignoreParseFailure(sessionsApi.authProviders({ mode: "login", authType: "oauth", machineId })), ignoreParseFailure(sessionsApi.saveApiKey("openai", "key", machineId)), ignoreParseFailure(sessionsApi.logoutProvider("openai", machineId)), @@ -94,7 +95,7 @@ describe("federated route contract", () => { vi.stubGlobal("WebSocket", FakeWebSocket); vi.stubGlobal("location", { protocol: "https:", host: "pi.example.test" }); - sessionEvents("s 1", machineId); + sessionEvents(session, machineId); globalSessionEvents(machineId); realtimeEvents(machineId); terminalSocket("p 1", "w 1", "t 1", { cols: 120, rows: 40 }, machineId); diff --git a/src/client/src/api/sockets.test.ts b/src/client/src/api/sockets.test.ts index 93b47ec..7942708 100644 --- a/src/client/src/api/sockets.test.ts +++ b/src/client/src/api/sockets.test.ts @@ -19,12 +19,12 @@ afterEach(() => { describe("machine-scoped socket urls", () => { it("defaults session sockets to the local machine scope", () => { - sessionEvents("s1"); + sessionEvents({ id: "s1", cwd: "/repo" }); globalSessionEvents(); realtimeEvents(); expect(webSocketUrls).toEqual([ - "wss://pi.example.test/api/machines/local/sessions/s1/events", + "wss://pi.example.test/api/machines/local/sessions/s1/events?cwd=%2Frepo", "wss://pi.example.test/api/machines/local/sessions/events", "wss://pi.example.test/api/machines/local/events", ]); diff --git a/src/client/src/api/sockets.ts b/src/client/src/api/sockets.ts index 40a6214..f97684d 100644 --- a/src/client/src/api/sockets.ts +++ b/src/client/src/api/sockets.ts @@ -1,5 +1,8 @@ -export function sessionEvents(sessionId: string, machineId = "local"): WebSocket { - return new WebSocket(`${webSocketBaseUrl()}${machinePrefix(machineId)}/sessions/${sessionId}/events`); +import type { SessionRef } from "../../../shared/apiTypes"; + +export function sessionEvents(session: SessionRef, machineId = "local"): WebSocket { + const query = new URLSearchParams({ cwd: session.cwd }).toString(); + return new WebSocket(`${webSocketBaseUrl()}${machinePrefix(machineId)}/sessions/${encodeURIComponent(session.id)}/events?${query}`); } export function globalSessionEvents(machineId = "local"): WebSocket { diff --git a/src/client/src/api/urls.ts b/src/client/src/api/urls.ts index 8201ac6..90f8282 100644 --- a/src/client/src/api/urls.ts +++ b/src/client/src/api/urls.ts @@ -1,3 +1,5 @@ +import type { SessionRef } from "../../../shared/apiTypes"; + export function gitDiffUrl(projectId: string, workspaceId: string, options?: { path?: string; staged?: boolean }): string { const params = new URLSearchParams(); if (options?.path !== undefined) params.set("path", options.path); @@ -14,12 +16,11 @@ export function machineGitDiffUrl(machineId: string, projectId: string, workspac return `/api/machines/${encodeURIComponent(machineId)}/projects/${encodeURIComponent(projectId)}/workspaces/${encodeURIComponent(workspaceId)}/git/diff${query ? `?${query}` : ""}`; } -export function messageUrl(sessionId: string, options?: { limit?: number; before?: number }, machineId = "local"): string { - const params = new URLSearchParams(); +export function messageUrl(session: SessionRef, options?: { limit?: number; before?: number }, machineId = "local"): string { + const params = new URLSearchParams({ cwd: session.cwd }); if (options?.limit !== undefined) params.set("limit", String(options.limit)); if (options?.before !== undefined) params.set("before", String(options.before)); - const query = params.toString(); - return `/api/machines/${encodeURIComponent(machineId)}/sessions/${sessionId}/messages${query ? `?${query}` : ""}`; + return `/api/machines/${encodeURIComponent(machineId)}/sessions/${encodeURIComponent(session.id)}/messages?${params.toString()}`; } export function workspaceImagePreviewUrl(projectId: string, workspaceId: string, path: string, options?: { modifiedAt?: string; machineId?: string }): string { diff --git a/src/client/src/components/PromptEditor.ts b/src/client/src/components/PromptEditor.ts index ea7b40f..81d0054 100644 --- a/src/client/src/components/PromptEditor.ts +++ b/src/client/src/components/PromptEditor.ts @@ -173,8 +173,8 @@ export class PromptEditor extends LitElement { this.completions = []; return; } - if (trigger.kind === "command" && this.sessionId !== undefined && this.sessionId !== "") { - const commands = await api.commands(this.sessionId, this.machineId).catch(emptySlashCommands); + if (trigger.kind === "command" && this.sessionId !== undefined && this.sessionId !== "" && this.cwd !== undefined && this.cwd !== "") { + const commands = await api.commands({ id: this.sessionId, cwd: this.cwd }, this.machineId).catch(emptySlashCommands); if (version !== this.requestVersion) return; this.completions = commands .filter((command) => command.name.toLowerCase().includes(trigger.query.toLowerCase())) diff --git a/src/client/src/controllers/authController.ts b/src/client/src/controllers/authController.ts index 3e39b8f..fe1b209 100644 --- a/src/client/src/controllers/authController.ts +++ b/src/client/src/controllers/authController.ts @@ -244,19 +244,19 @@ export class AuthController { } private async refreshStatus(): Promise { - const sessionId = this.sessionId(); - if (sessionId === undefined) return; + const session = this.session(); + if (session === undefined) return; try { - this.applyStatus(await this.api.status(sessionId, selectedMachineId(this.getState()))); + this.applyStatus(await this.api.status(session, selectedMachineId(this.getState()))); } catch { // Status refresh is opportunistic after login completes. } } - private sessionId(): string | undefined { + private session() { const session = this.getState().selectedSession; if (session === undefined || session.archived === true) return undefined; - return session.id; + return session; } } diff --git a/src/client/src/controllers/sessionController.test.ts b/src/client/src/controllers/sessionController.test.ts index eead33c..01d3bc6 100644 --- a/src/client/src/controllers/sessionController.test.ts +++ b/src/client/src/controllers/sessionController.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, it } from "vitest"; -import { api as defaultApi, type MessagePage, type SessionActivity, type SessionInfo, type SessionStatus, type Workspace } from "../api"; +import { api as defaultApi, type MessagePage, type SessionActivity, type SessionInfo, type SessionRef, type SessionStatus, type Workspace } from "../api"; import { isCachedNewSessionInfo, loadCachedNewSessions, markCachedNewSessionInfo, rememberCachedNewSession } from "../cachedNewSessions"; import { initialAppState, type AppState } from "../appState"; import { machineSessionKey } from "../machineKeys"; @@ -38,8 +38,8 @@ class MemoryStorage implements Storage { class FakeSocket implements SessionEventSocket { readonly connectedSessionIds: string[] = []; - connect(sessionId: string): void { - this.connectedSessionIds.push(sessionId); + connect(session: SessionRef): void { + this.connectedSessionIds.push(session.id); } setHandler(): void { @@ -179,11 +179,11 @@ describe("SessionController", () => { const api: typeof defaultApi = { ...defaultApi, startSession: () => Promise.resolve(replacementSession), - messages: (sessionId) => { - if (sessionId === oldSession.id) return Promise.reject(new Error("Session not found")); + messages: (session) => { + if (session.id === oldSession.id) return Promise.reject(new Error("Session not found")); return Promise.resolve(emptyPage); }, - status: (sessionId) => Promise.resolve(status(sessionId)), + status: (session) => Promise.resolve(status(session.id)), }; const controller = new SessionController( () => state, @@ -220,7 +220,7 @@ describe("SessionController", () => { ...defaultApi, respondToCommand: () => Promise.resolve({ type: "done", message: "Session forked", session: replacementSession, promptDraft: "fork me" }), messages: () => Promise.resolve(emptyPage), - status: (sessionId) => Promise.resolve(status(sessionId)), + status: (session) => Promise.resolve(status(session.id)), }; const controller = new SessionController( () => state, @@ -243,7 +243,7 @@ describe("SessionController", () => { ...defaultApi, archive: () => Promise.resolve({ archived: true }), messages: () => Promise.resolve(emptyPage), - status: (sessionId) => Promise.resolve(status(sessionId)), + status: (session) => Promise.resolve(status(session.id)), }; const controller = new SessionController( () => state, @@ -272,7 +272,7 @@ describe("SessionController", () => { ...defaultApi, archiveWithDescendants: () => Promise.resolve({ archived: true, sessionIds: [oldSession.id, childSession.id], archivedCount: 2, skippedAlreadyArchivedCount: 0 }), messages: () => Promise.resolve(emptyPage), - status: (sessionId) => Promise.resolve(status(sessionId)), + status: (session) => Promise.resolve(status(session.id)), }; const controller = new SessionController( () => state, diff --git a/src/client/src/controllers/sessionController.ts b/src/client/src/controllers/sessionController.ts index 3bc4bba..486c7db 100644 --- a/src/client/src/controllers/sessionController.ts +++ b/src/client/src/controllers/sessionController.ts @@ -1,4 +1,4 @@ -import { api as defaultApi, type CommandResult, type SessionActivity, type SessionInfo, type SessionStatus, type ThinkingLevel } from "../api"; +import { api as defaultApi, type CommandResult, type SessionActivity, type SessionInfo, type SessionRef, type SessionStatus, type ThinkingLevel } from "../api"; import type { AppState } from "../appState"; import { forgetCachedNewSession, isCachedNewSessionInfo, markCachedNewSessionInfo, rememberCachedNewSession, stripCachedNewSessionMarker } from "../cachedNewSessions"; import { textMessage } from "../chatMessages"; @@ -14,7 +14,7 @@ import { selectedMachineId, type GetState, type SetState, type UpdateUrl } from const MESSAGE_PAGE_SIZE = 100; export interface SessionEventSocket { - connect(sessionId: string, onEvent: (event: SessionUiEvent) => void, onReconnect?: () => void, machineId?: string): void; + connect(session: SessionRef, onEvent: (event: SessionUiEvent) => void, onReconnect?: () => void, machineId?: string): void; setHandler(onEvent: (event: SessionUiEvent) => void): void; close(): void; } @@ -116,7 +116,7 @@ export class SessionController { }); try { if (session.archived === true) { - const page = await this.api.messages(session.id, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())); + const page = await this.api.messages(session, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())); if (seq !== this.selectionSeq || this.getState().selectedSession?.id !== session.id) return; const history = this.transcripts.mergeHistory(transcriptKey, page); this.setState({ ...history, isLoadingEarlierMessages: false, isReceivingPartialStream: false, status: undefined, activity: undefined }); @@ -125,12 +125,12 @@ export class SessionController { } const buffered: SessionUiEvent[] = []; this.socket.connect( - session.id, + session, (event) => buffered.push(event), () => { void this.refreshSelectedSession(session.id); }, selectedMachineId(this.getState()), ); - const [page, status] = await Promise.all([this.api.messages(session.id, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())), this.api.status(session.id, selectedMachineId(this.getState()))]); + const [page, status] = await Promise.all([this.api.messages(session, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())), this.api.status(session, selectedMachineId(this.getState()))]); if (seq !== this.selectionSeq || this.getState().selectedSession?.id !== session.id) return; const history = this.transcripts.mergeHistory(transcriptKey, page); const isReceivingPartialStream = status.isStreaming; @@ -156,7 +156,7 @@ export class SessionController { if (!session || state.isLoadingEarlierMessages || state.messagePageStart <= 0) return; this.setState({ isLoadingEarlierMessages: true }); try { - const page = await this.api.messages(session.id, { before: state.messagePageStart, limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())); + const page = await this.api.messages(session, { before: state.messagePageStart, limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())); if (this.getState().selectedSession?.id !== session.id) return; const history = this.transcripts.mergeHistory(this.sessionCacheKey(session.id), page); this.setState(history); @@ -174,7 +174,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session || session.archived === true) return; try { - await this.api.prompt(session.id, text, streamingBehavior, selectedMachineId(this.getState())); + await this.api.prompt(session, text, streamingBehavior, selectedMachineId(this.getState())); this.markCachedNewSessionPersisted(session); } catch (error) { this.setState({ error: String(error) }); @@ -186,7 +186,7 @@ export class SessionController { if (!session || session.archived === true) return; this.setState({ messages: [...this.getState().messages, textMessage("user", text)] }); try { - await this.api.shell(session.id, text, selectedMachineId(this.getState())); + await this.api.shell(session, text, selectedMachineId(this.getState())); this.markCachedNewSessionPersisted(session); } catch (error) { this.setState({ messages: [...this.getState().messages, textMessage("system", String(error))], error: String(error) }); @@ -198,7 +198,7 @@ export class SessionController { if (!session || session.archived === true) return; this.setState({ messages: [...this.getState().messages, textMessage("user", text)] }); try { - this.applyCommandResult(await this.api.runCommand(session.id, text, selectedMachineId(this.getState()))); + this.applyCommandResult(await this.api.runCommand(session, text, selectedMachineId(this.getState()))); this.markCachedNewSessionPersisted(session); } catch (error) { this.setState({ messages: [...this.getState().messages, textMessage("system", String(error))], error: String(error) }); @@ -210,7 +210,7 @@ export class SessionController { if (!session) return; this.setState({ commandDialog: undefined }); try { - this.applyCommandResult(await this.api.respondToCommand(session.id, requestId, value, selectedMachineId(this.getState()))); + this.applyCommandResult(await this.api.respondToCommand(session, requestId, value, selectedMachineId(this.getState()))); } catch (error) { this.setState({ error: String(error) }); } @@ -231,7 +231,7 @@ export class SessionController { return; } try { - await this.api.archive(session.id, selectedMachineId(this.getState())); + await this.api.archive(session, selectedMachineId(this.getState())); const state = this.getState(); const sessions = markSessionArchived(state.sessions, session.id, new Date().toISOString()); const selectionChange = selectionAfterArchivingSession(sessions, state.selectedSession?.id, session.id); @@ -247,7 +247,7 @@ export class SessionController { async archiveSessionWithDescendants(session = this.getState().selectedSession) { if (!session || isCachedNewSessionInfo(session)) return; try { - const response = await this.api.archiveWithDescendants(session.id, selectedMachineId(this.getState())); + const response = await this.api.archiveWithDescendants(session, selectedMachineId(this.getState())); const archivedIds = response.sessionIds !== undefined && response.sessionIds.length > 0 ? response.sessionIds : [session.id]; const state = this.getState(); const sessions = markSessionsArchived(state.sessions, archivedIds, new Date().toISOString()); @@ -263,7 +263,7 @@ export class SessionController { async deleteCachedNewSession(session = this.getState().selectedSession) { if (!isCachedNewSessionInfo(session)) return; - void this.api.stop(session.id, selectedMachineId(this.getState())).catch(() => { + void this.api.stop(session, selectedMachineId(this.getState())).catch(() => { // Best-effort cleanup for browser-cached sessions that may not exist server-side anymore. }); forgetCachedNewSession(session.id, selectedMachineId(this.getState())); @@ -282,7 +282,7 @@ export class SessionController { async restoreSession(session = this.getState().selectedSession) { if (!session) return; try { - await this.api.restore(session.id, selectedMachineId(this.getState())); + await this.api.restore(session, selectedMachineId(this.getState())); const restored = { ...session }; delete restored.archived; delete restored.archivedAt; @@ -296,7 +296,7 @@ export class SessionController { async detachParent(session = this.getState().selectedSession) { if (session?.parentSessionPath === undefined) return; try { - await this.api.detachParent(session.id, selectedMachineId(this.getState())); + await this.api.detachParent(session, selectedMachineId(this.getState())); const detached = { ...session }; delete detached.parentSessionPath; this.replaceSession(detached); @@ -309,7 +309,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session || session.archived === true) return []; try { - return (await this.api.models(session.id, selectedMachineId(this.getState()))).models; + return (await this.api.models(session, selectedMachineId(this.getState()))).models; } catch (error) { this.setState({ error: String(error) }); return []; @@ -320,7 +320,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session || session.archived === true) return; try { - this.applyStatus(await this.api.setModel(session.id, provider, modelId, selectedMachineId(this.getState()))); + this.applyStatus(await this.api.setModel(session, provider, modelId, selectedMachineId(this.getState()))); } catch (error) { this.setState({ error: String(error) }); } @@ -330,7 +330,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session || session.archived === true) return; try { - this.applyStatus(await this.api.cycleModel(session.id, direction, selectedMachineId(this.getState()))); + this.applyStatus(await this.api.cycleModel(session, direction, selectedMachineId(this.getState()))); } catch (error) { this.setState({ error: String(error) }); } @@ -340,7 +340,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session || session.archived === true) return []; try { - return (await this.api.thinkingLevels(session.id, selectedMachineId(this.getState()))).levels; + return (await this.api.thinkingLevels(session, selectedMachineId(this.getState()))).levels; } catch (error) { this.setState({ error: String(error) }); return []; @@ -351,7 +351,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session || session.archived === true) return; try { - this.applyStatus(await this.api.setThinkingLevel(session.id, level, selectedMachineId(this.getState()))); + this.applyStatus(await this.api.setThinkingLevel(session, level, selectedMachineId(this.getState()))); } catch (error) { this.setState({ error: String(error) }); } @@ -361,7 +361,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session || session.archived === true) return; try { - this.applyStatus(await this.api.cycleThinkingLevel(session.id, selectedMachineId(this.getState()))); + this.applyStatus(await this.api.cycleThinkingLevel(session, selectedMachineId(this.getState()))); } catch (error) { this.setState({ error: String(error) }); } @@ -371,7 +371,7 @@ export class SessionController { const session = this.getState().selectedSession; if (!session) return; try { - await this.api.abort(session.id, selectedMachineId(this.getState())); + await this.api.abort(session, selectedMachineId(this.getState())); } catch (error) { this.setState({ error: String(error) }); } @@ -382,7 +382,7 @@ export class SessionController { if (sessionId === undefined || session?.id !== sessionId || session.archived === true) return; try { this.flushPendingTranscriptEvents(); - const [page, status] = await Promise.all([this.api.messages(sessionId, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())), this.api.status(sessionId, selectedMachineId(this.getState()))]); + const [page, status] = await Promise.all([this.api.messages(session, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())), this.api.status(session, selectedMachineId(this.getState()))]); if (this.getState().selectedSession?.id !== sessionId) return; const history = this.transcripts.mergeHistory(this.sessionCacheKey(sessionId), page); this.setState({ @@ -548,7 +548,9 @@ export class SessionController { private async refreshMessages(sessionId: string) { try { - const page = await this.api.messages(sessionId, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())); + const session = this.getState().selectedSession; + if (session?.id !== sessionId) return; + const page = await this.api.messages(session, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())); if (this.getState().selectedSession?.id !== sessionId) return; this.setState(this.transcripts.mergeHistory(this.sessionCacheKey(sessionId), page)); } catch (error) { diff --git a/src/client/src/sessionSocket.ts b/src/client/src/sessionSocket.ts index c07fc56..e68f839 100644 --- a/src/client/src/sessionSocket.ts +++ b/src/client/src/sessionSocket.ts @@ -1,11 +1,11 @@ import { globalSessionEvents, realtimeEvents, sessionEvents } from "./api"; -import type { GlobalSessionEvent, RealtimeEvent, SessionUiEvent } from "../../shared/apiTypes"; +import type { GlobalSessionEvent, RealtimeEvent, SessionRef, SessionUiEvent } from "../../shared/apiTypes"; export type { GlobalSessionEvent, RealtimeEvent, SessionUiEvent } from "../../shared/apiTypes"; export class SessionSocket { private socket: WebSocket | undefined; - private sessionId: string | undefined; + private session: SessionRef | undefined; private onEvent: ((event: SessionUiEvent) => void) | undefined; private reconnectTimer?: number; private reconnectDelay = 500; @@ -14,10 +14,10 @@ export class SessionSocket { private onReconnect: (() => void) | undefined; private machineId = "local"; - connect(sessionId: string, onEvent: (event: SessionUiEvent) => void, onReconnect?: () => void, machineId = "local"): void { + connect(session: SessionRef, onEvent: (event: SessionUiEvent) => void, onReconnect?: () => void, machineId = "local"): void { this.close(); this.machineId = machineId; - this.sessionId = sessionId; + this.session = session; this.onEvent = onEvent; this.onReconnect = onReconnect; this.shouldReconnect = true; @@ -33,7 +33,7 @@ export class SessionSocket { window.clearTimeout(this.reconnectTimer); closeSocketQuietly(this.socket); this.socket = undefined; - this.sessionId = undefined; + this.session = undefined; this.onEvent = undefined; this.onReconnect = undefined; this.hasOpened = false; @@ -41,8 +41,8 @@ export class SessionSocket { } private open(): void { - if (this.sessionId === undefined || this.sessionId === "" || !this.shouldReconnect) return; - const socket = sessionEvents(this.sessionId, this.machineId); + if (this.session === undefined || this.session.id === "" || this.session.cwd === "" || !this.shouldReconnect) return; + const socket = sessionEvents(this.session, this.machineId); this.socket = socket; socket.onopen = () => { this.reconnectDelay = 500; diff --git a/src/server/sessiond/sessionProxyRoutes.ts b/src/server/sessiond/sessionProxyRoutes.ts index 68b1f55..ef87aa9 100644 --- a/src/server/sessiond/sessionProxyRoutes.ts +++ b/src/server/sessiond/sessionProxyRoutes.ts @@ -24,7 +24,7 @@ export function registerSessionProxyRoutes(app: FastifyInstance, daemon: Session app.get(`${prefix}/sessiond/health`, (_request, reply) => proxy({ method: "GET", url: `${prefix}/health` }, reply)); app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/events`, { websocket: true }, (socket, request) => { - bridgeSockets(socket, daemon.connectWebSocket(`/sessions/${request.params.sessionId}/events`)); + bridgeSockets(socket, daemon.connectWebSocket(stripPrefix(request.url, prefix))); }); app.get(`${prefix}/sessions/events`, { websocket: true }, (socket) => { diff --git a/src/server/sessions/piSessionManagerGateway.test.ts b/src/server/sessions/piSessionManagerGateway.test.ts new file mode 100644 index 0000000..d5f80f7 --- /dev/null +++ b/src/server/sessions/piSessionManagerGateway.test.ts @@ -0,0 +1,85 @@ +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { createPiSessionManagerGateway, defaultPiSessionDir, defaultPiSessionsRoot, SessionDirResolver } from "./piSessionManagerGateway.js"; +import type { PiSessionManager } from "./piSessionService.js"; + +let tempDir: string; +let agentDir: string; +let cwd: string; + +beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), "pi-web-session-gateway-test-")); + agentDir = join(tempDir, "agent"); + cwd = join(tempDir, "workspace"); + await mkdir(cwd, { recursive: true }); +}); + +afterEach(async () => { + await rm(tempDir, { recursive: true, force: true }); +}); + +describe("SessionDirResolver", () => { + it("uses Pi default session storage when no Pi override is configured", () => { + const resolver = new SessionDirResolver({ agentDir, env: {} }); + + expect(resolver.resolve(cwd)).toMatchObject({ source: "pi-default", sessionDir: defaultPiSessionDir(cwd, agentDir), usesConfiguredSessionDir: false }); + expect(defaultPiSessionsRoot(agentDir)).toBe(join(agentDir, "sessions")); + }); + + it("uses Pi sessionDir settings and resolves relative paths against the session cwd", async () => { + await mkdir(agentDir, { recursive: true }); + await writeFile(join(agentDir, "settings.json"), `${JSON.stringify({ sessionDir: ".pi/sessions" }, null, 2)}\n`, "utf8"); + + const resolver = new SessionDirResolver({ agentDir, env: {} }); + + expect(resolver.resolve(cwd)).toMatchObject({ source: "settings", sessionDir: join(cwd, ".pi", "sessions"), usesConfiguredSessionDir: true }); + }); + + it("lets project-local Pi sessionDir settings override global Pi settings for that cwd", async () => { + await mkdir(agentDir, { recursive: true }); + await mkdir(join(cwd, ".pi"), { recursive: true }); + await writeFile(join(agentDir, "settings.json"), `${JSON.stringify({ sessionDir: join(tempDir, "global-sessions") }, null, 2)}\n`, "utf8"); + await writeFile(join(cwd, ".pi", "settings.json"), `${JSON.stringify({ sessionDir: ".workspace-sessions" }, null, 2)}\n`, "utf8"); + + const resolver = new SessionDirResolver({ agentDir, env: {} }); + + expect(resolver.resolve(cwd)).toMatchObject({ source: "settings", sessionDir: join(cwd, ".workspace-sessions"), usesConfiguredSessionDir: true }); + }); + + it("lets the Pi sessionDir environment override Pi settings", async () => { + const envDir = join(tempDir, "env-sessions"); + await mkdir(agentDir, { recursive: true }); + await writeFile(join(agentDir, "settings.json"), `${JSON.stringify({ sessionDir: join(tempDir, "settings-sessions") }, null, 2)}\n`, "utf8"); + + const resolver = new SessionDirResolver({ agentDir, env: { PI_CODING_AGENT_SESSION_DIR: envDir } }); + + expect(resolver.resolve(cwd)).toMatchObject({ source: "env", sessionDir: envDir, usesConfiguredSessionDir: true }); + }); +}); + +describe("Pi session manager gateway", () => { + it("lists only sessions for the requested cwd when a custom Pi sessionDir is shared", async () => { + const sharedSessionDir = join(tempDir, "shared-sessions"); + const otherCwd = join(tempDir, "other-workspace"); + await writeSessionFile(sharedSessionDir, "session-a", cwd); + await writeSessionFile(sharedSessionDir, "session-b", otherCwd); + const gateway = createPiSessionManagerGateway({ agentDir, env: { PI_CODING_AGENT_SESSION_DIR: sharedSessionDir } }); + + await expect(gateway.list(cwd)).resolves.toMatchObject([{ id: "session-a", cwd }]); + const created = gateway.create(cwd); + expect(hasSessionDir(created)).toBe(true); + if (!hasSessionDir(created)) throw new Error("Expected SDK session manager"); + expect(created.getSessionDir()).toBe(sharedSessionDir); + }); +}); + +function hasSessionDir(manager: PiSessionManager): manager is PiSessionManager & { getSessionDir(): string } { + return "getSessionDir" in manager && typeof manager.getSessionDir === "function"; +} + +async function writeSessionFile(dir: string, id: string, sessionCwd: string): Promise { + await mkdir(dir, { recursive: true }); + await writeFile(join(dir, `${id}.jsonl`), `${JSON.stringify({ type: "session", version: 3, id, timestamp: "2026-01-01T00:00:00.000Z", cwd: sessionCwd })}\n`, "utf8"); +} diff --git a/src/server/sessions/piSessionManagerGateway.ts b/src/server/sessions/piSessionManagerGateway.ts new file mode 100644 index 0000000..2aabd99 --- /dev/null +++ b/src/server/sessions/piSessionManagerGateway.ts @@ -0,0 +1,100 @@ +import { homedir } from "node:os"; +import { dirname, isAbsolute, join, resolve } from "node:path"; +import { getAgentDir, SessionManager, SettingsManager } from "@earendil-works/pi-coding-agent"; +import type { PiSessionListEntry, PiSessionManager, PiSessionManagerGateway } from "./piSessionService.js"; + +export const PI_SESSION_DIR_ENV = "PI_CODING_AGENT_SESSION_DIR"; + +type SessionDirSource = "env" | "settings" | "pi-default"; + +export interface SessionDirResolution { + source: SessionDirSource; + sessionDir: string; + usesConfiguredSessionDir: boolean; +} + +export interface SessionDirResolverOptions { + agentDir?: string; + env?: NodeJS.ProcessEnv; +} + +export class SessionDirResolver { + private readonly agentDir: string; + private readonly env: NodeJS.ProcessEnv; + + constructor(options: SessionDirResolverOptions = {}) { + this.agentDir = options.agentDir ?? getAgentDir(); + this.env = options.env ?? process.env; + } + + resolve(cwd: string): SessionDirResolution { + const envSessionDir = this.env[PI_SESSION_DIR_ENV]; + if (envSessionDir !== undefined && envSessionDir !== "") { + return { source: "env", sessionDir: resolveConfiguredPath(envSessionDir, cwd), usesConfiguredSessionDir: true }; + } + + const settingsSessionDir = SettingsManager.create(cwd, this.agentDir).getSessionDir(); + if (settingsSessionDir !== undefined && settingsSessionDir !== "") { + return { source: "settings", sessionDir: resolveConfiguredPath(settingsSessionDir, cwd), usesConfiguredSessionDir: true }; + } + + return { source: "pi-default", sessionDir: defaultPiSessionDir(cwd, this.agentDir), usesConfiguredSessionDir: false }; + } +} + +export type PiSessionManagerGatewayOptions = SessionDirResolverOptions; + +export function createPiSessionManagerGateway(options: PiSessionManagerGatewayOptions = {}): PiSessionManagerGateway { + return new SettingsAwarePiSessionManagerGateway(new SessionDirResolver(options)); +} + +class SettingsAwarePiSessionManagerGateway implements PiSessionManagerGateway { + constructor(private readonly resolver: SessionDirResolver) {} + + async list(cwd: string): Promise { + const resolution = this.resolver.resolve(cwd); + return filterSessionsForCwd(await listSessionsInDir(resolution.sessionDir), cwd); + } + + create(cwd: string): PiSessionManager { + const resolution = this.resolver.resolve(cwd); + return SessionManager.create(cwd, resolution.sessionDir); + } + + open(path: string): PiSessionManager { + return SessionManager.open(path, dirname(path)); + } +} + +export async function listSessionsInDir(sessionDir: string): Promise { + return SessionManager.list("", sessionDir); +} + +export function filterSessionsForCwd(sessions: readonly PiSessionListEntry[], cwd: string): PiSessionListEntry[] { + return sessions.filter((session) => session.cwd === cwd); +} + +export function defaultPiSessionsRoot(agentDir = getAgentDir()): string { + return join(agentDir, "sessions"); +} + +export function defaultPiSessionDir(cwd: string, agentDir = getAgentDir()): string { + return sessionDirInDefaultPiStore(defaultPiSessionsRoot(agentDir), cwd); +} + +export function sessionDirInDefaultPiStore(storeRoot: string, cwd: string): string { + const safePath = `--${cwd.replace(/^[/\\]/u, "").replace(/[/\\:]/gu, "-")}--`; + return join(storeRoot, safePath); +} + +export function resolveConfiguredPath(path: string, cwd: string): string { + const expanded = expandTildePath(path); + return isAbsolute(expanded) ? expanded : resolve(cwd, expanded); +} + +function expandTildePath(path: string): string { + if (path === "~") return homedir(); + if (path.startsWith("~/")) return join(homedir(), path.slice(2)); + return path; +} + diff --git a/src/server/sessions/piSessionService.test.ts b/src/server/sessions/piSessionService.test.ts index eb23fc1..4adbdec 100644 --- a/src/server/sessions/piSessionService.test.ts +++ b/src/server/sessions/piSessionService.test.ts @@ -43,6 +43,10 @@ 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: "" }; } +function sessionRef(id: string, cwd = "/workspace") { + return { id, cwd }; +} + function fakeRuntime(sessionId = "session-1", patch: Partial = {}) { const promptCalls: { text: string; options: unknown }[] = []; const listeners: ((event: unknown) => void)[] = []; @@ -122,7 +126,6 @@ function sessionGateway(records: ReturnType[]): SessionGat return { create: () => fakeSessionManager(), list: () => Promise.resolve(records), - listAll: () => Promise.resolve(records), open: () => fakeSessionManager(), }; } @@ -174,7 +177,7 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 1_000, }); - await service.status("idle-session"); + await service.status(sessionRef("idle-session")); hub.globalEvents.length = 0; listener?.({ type: "agent_start" }); @@ -209,7 +212,7 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.status("completion-session"); + await service.status(sessionRef("completion-session")); hub.globalEvents.length = 0; listener?.({ type: "tool_execution_end", toolName: "read", isError: false }); @@ -235,7 +238,6 @@ describe("PiSessionService", () => { { ...sessionRecord("active"), messageCount: 1, firstMessage: "hello", allMessagesText: "hello" }, { ...sessionRecord("archived"), messageCount: 2, firstMessage: "bye", allMessagesText: "bye" }, ]), - listAll: () => Promise.resolve([]), open: () => fakeSessionManager(), }, heartbeatIntervalMs: 60_000, @@ -262,7 +264,6 @@ describe("PiSessionService", () => { sessionManager: { create: () => fakeSessionManager(), list: () => Promise.resolve([{ ...sessionRecord("active"), messageCount: 1, firstMessage: "hello", allMessagesText: "hello" }]), - listAll: () => Promise.resolve([]), open: () => fakeSessionManager(), }, heartbeatIntervalMs: 60_000, @@ -301,13 +302,12 @@ describe("PiSessionService", () => { sessionManager: { create: () => fakeSessionManager(), list: (cwd) => Promise.resolve(cwd === "/workspace" ? [root, directChild, archivedChild, grandchild] : [otherWorkspaceChild]), - listAll: () => Promise.resolve([root, directChild, archivedChild, grandchild, otherWorkspaceChild]), open: () => fakeSessionManager(), }, heartbeatIntervalMs: 60_000, }); - await expect(service.archiveTree("root")).resolves.toEqual({ + await expect(service.archiveTree(sessionRef("root"))).resolves.toEqual({ archived: true, sessionIds: ["root", "direct-child", "grandchild"], archivedCount: 3, @@ -331,7 +331,6 @@ describe("PiSessionService", () => { sessionManager: { create: () => fakeSessionManager(), list: () => Promise.resolve([]), - listAll: () => Promise.resolve([]), open: () => fakeSessionManager(), }, workspaceActivity: { @@ -360,7 +359,7 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.prompt("prompt-session", "Build the thing"); + await service.prompt(sessionRef("prompt-session"), "Build the thing"); expect(fake.calls.prompt).toEqual([{ text: "Build the thing", options: undefined }]); await service.dispose(); @@ -379,7 +378,7 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await expect(service.status("status-session")).resolves.toMatchObject({ + 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, @@ -399,7 +398,7 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.prompt("dedupe-session", "already queued", "followUp"); + await service.prompt(sessionRef("dedupe-session"), "already queued", "followUp"); expect(fake.calls.prompt).toEqual([]); await service.dispose(); @@ -414,7 +413,7 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.prompt("queued-session", "Wait for the current turn", "followUp"); + 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); @@ -439,12 +438,12 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.prompt("compacting-session", "Start task 1", "followUp"); - await service.prompt("compacting-session", "Then task 2", "followUp"); + 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("compacting-session")).resolves.toMatchObject({ + await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({ pendingMessageCount: 2, queuedMessages: [{ kind: "followUp", text: "Start task 1" }, { kind: "followUp", text: "Then task 2" }], }); @@ -455,7 +454,7 @@ describe("PiSessionService", () => { 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("compacting-session")).resolves.toMatchObject({ + await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({ pendingMessageCount: 1, queuedMessages: [{ kind: "followUp", text: "Then task 2" }], }); @@ -467,7 +466,7 @@ describe("PiSessionService", () => { { text: "Start task 1", options: undefined }, { text: "Then task 2", options: { streamingBehavior: "followUp" } }, ]); - await expect(service.status("compacting-session")).resolves.toMatchObject({ + await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({ pendingMessageCount: 0, queuedMessages: [], }); @@ -483,8 +482,8 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.status("abort-session"); - await service.abort("abort-session"); + await service.status(sessionRef("abort-session")); + await service.abort(sessionRef("abort-session")); expect(fake.calls.clearQueue).toBe(1); expect(fake.calls.abort).toBe(1); @@ -499,13 +498,13 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.prompt("abort-compaction-session", "Do not deliver after abort", "followUp"); - await expect(service.status("abort-compaction-session")).resolves.toMatchObject({ pendingMessageCount: 1 }); - await service.abort("abort-compaction-session"); + 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("abort-compaction-session")).resolves.toMatchObject({ pendingMessageCount: 0, queuedMessages: [] }); + await expect(service.status(sessionRef("abort-compaction-session"))).resolves.toMatchObject({ pendingMessageCount: 0, queuedMessages: [] }); await service.dispose(); }); @@ -524,7 +523,7 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.status("auth-session"); + await service.status(sessionRef("auth-session")); hub.sessionEvents.length = 0; hub.globalEvents.length = 0; @@ -553,8 +552,8 @@ describe("PiSessionService", () => { heartbeatIntervalMs: 60_000, }); - await service.status("stop-session"); - service.stop("stop-session"); + await service.status(sessionRef("stop-session")); + service.stop(sessionRef("stop-session")); expect(fake.calls.clearQueue).toBe(1); await service.dispose(); diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index cdb29ed..36f7171 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -13,7 +13,7 @@ import { type CreateAgentSessionRuntimeFactory, type EditToolDetails, } from "@earendil-works/pi-coding-agent"; -import type { ClientArchiveSessionsResponse, ClientCommand, ClientCommandResult, ClientMessagePage, ClientSession, ClientSessionModel, ClientSessionStatus, ClientThinkingLevel, SessionUiEvent } from "../types.js"; +import type { ClientArchiveSessionsResponse, ClientCommand, ClientCommandResult, ClientMessagePage, ClientSession, ClientSessionModel, ClientSessionRef, ClientSessionStatus, ClientThinkingLevel, SessionUiEvent } from "../types.js"; import { pageMessagesAtSafeBoundary } from "./messagePaging.js"; import type { SessionEventHub } from "../realtime/sessionEventHub.js"; import { BUILTIN_COMMANDS } from "./builtinCommands.js"; @@ -24,6 +24,7 @@ import type { ActiveSession } from "./sessionRuntimeStore.js"; import type { AuthChange } from "./authService.js"; import { fallbackSessionName, generateShortSessionName } from "./sessionNameGenerator.js"; import { computeEditPreview, type EditPreviewResult } from "./editPreview.js"; +import { createPiSessionManagerGateway } from "./piSessionManagerGateway.js"; import type { WorkspaceActivityService } from "../activity/workspaceActivityService.js"; function noop(): void { @@ -34,6 +35,18 @@ function authLossWarningKey(sessionId: string, provider: string, modelId: string return `${sessionId}:${provider}/${modelId}`; } +function sessionIdFromLookup(ref: PiSessionLookup): string { + return typeof ref === "string" ? ref : ref.id; +} + +function isPiSessionRef(ref: PiSessionLookup): ref is PiSessionRef { + return typeof ref !== "string"; +} + +function lookupMatchesActiveSession(ref: PiSessionLookup, active: ActiveSession): boolean { + return !isPiSessionRef(ref) || active.runtime.cwd === ref.cwd; +} + type QueuedPromptKind = "steer" | "followUp"; interface QueuedPrompt { @@ -42,7 +55,12 @@ interface QueuedPrompt { } type SessionArchiveRepository = Pick; -interface PiSessionListEntry { + +export type PiSessionRef = ClientSessionRef; + +type PiSessionLookup = string | PiSessionRef; + +export interface PiSessionListEntry { id: string; path: string; cwd: string; @@ -74,7 +92,6 @@ export interface PiSessionManager { export interface PiSessionManagerGateway { list(cwd: string): Promise; create(cwd: string): PiSessionManager; - listAll(): Promise; open(path: string): PiSessionManager; } @@ -201,7 +218,7 @@ export class PiSessionService { constructor(private readonly events: SessionEventHub, deps: PiSessionServiceDependencies = {}) { this.archiveStore = deps.archiveStore ?? new SessionArchiveStore(); this.agentDir = deps.agentDir ?? getAgentDir(); - this.sessionManager = deps.sessionManager ?? SessionManager; + this.sessionManager = deps.sessionManager ?? createPiSessionManagerGateway({ agentDir: this.agentDir }); this.modelRegistry = deps.modelRegistry ?? ModelRegistry.create(AuthStorage.create()); this.createRuntime = deps.createRuntime ?? createDefaultRuntimeFactory(this.modelRegistry.authStorage, this.modelRegistry); this.createAgentRuntime = deps.createAgentRuntime ?? defaultCreateAgentRuntime; @@ -277,17 +294,17 @@ export class PiSessionService { }; } - async messages(sessionId: string, page?: { before?: number; limit?: number }): Promise { - const session = await this.getOrOpen(sessionId); + async messages(ref: PiSessionLookup, page?: { before?: number; limit?: number }): Promise { + const session = await this.getOrOpen(ref); return pageMessagesAtSafeBoundary(historyMessages(session), page); } - async status(sessionId: string): Promise { - return this.statusFromSession(await this.getOrOpen(sessionId)); + async status(ref: PiSessionLookup): Promise { + return this.statusFromSession(await this.getOrOpen(ref)); } - async availableModels(sessionId: string): Promise { - const session = await this.getOrOpen(sessionId); + async availableModels(ref: PiSessionLookup): Promise { + const session = await this.getOrOpen(ref); session.modelRegistry.refresh(); const models = session.scopedModels.length > 0 ? session.scopedModels.map((scoped) => scoped.model) @@ -295,9 +312,9 @@ export class PiSessionService { return models.map(modelToClientModel); } - async setModel(sessionId: string, provider: string, modelId: string): Promise { - await this.assertWritable(sessionId); - const session = await this.getOrOpen(sessionId); + async setModel(ref: PiSessionLookup, provider: string, modelId: string): Promise { + await this.assertWritable(ref); + const session = await this.getOrOpen(ref); session.modelRegistry.refresh(); const candidates = session.scopedModels.length > 0 ? session.scopedModels.map((scoped) => scoped.model) @@ -311,9 +328,9 @@ export class PiSessionService { return this.statusFromSession(session); } - async cycleModel(sessionId: string, direction: "forward" | "backward"): Promise { - await this.assertWritable(sessionId); - const session = await this.getOrOpen(sessionId); + async cycleModel(ref: PiSessionLookup, direction: "forward" | "backward"): Promise { + await this.assertWritable(ref); + const session = await this.getOrOpen(ref); const result = await session.cycleModel(direction); if (result === undefined) throw new Error(session.scopedModels.length > 0 ? "Only one model in scope" : "Only one model available"); this.publishActivity(session, `model: ${result.model.id}`, "idle", result.model.provider); @@ -321,23 +338,23 @@ export class PiSessionService { return this.statusFromSession(session); } - async availableThinkingLevels(sessionId: string): Promise { - const session = await this.getOrOpen(sessionId); + async availableThinkingLevels(ref: PiSessionLookup): Promise { + const session = await this.getOrOpen(ref); return session.getAvailableThinkingLevels(); } - async setThinkingLevel(sessionId: string, level: ClientThinkingLevel): Promise { - await this.assertWritable(sessionId); - const session = await this.getOrOpen(sessionId); + async setThinkingLevel(ref: PiSessionLookup, level: ClientThinkingLevel): Promise { + await this.assertWritable(ref); + const session = await this.getOrOpen(ref); session.setThinkingLevel(level); this.publishActivity(session, `thinking: ${session.thinkingLevel}`, "idle"); this.publishStatus(session); return this.statusFromSession(session); } - async cycleThinkingLevel(sessionId: string): Promise { - await this.assertWritable(sessionId); - const session = await this.getOrOpen(sessionId); + async cycleThinkingLevel(ref: PiSessionLookup): Promise { + await this.assertWritable(ref); + const session = await this.getOrOpen(ref); const level = session.cycleThinkingLevel(); if (level === undefined) throw new Error("Current model does not support thinking"); this.publishActivity(session, `thinking: ${level}`, "idle"); @@ -345,8 +362,8 @@ export class PiSessionService { return this.statusFromSession(session); } - async commands(sessionId: string): Promise { - const session = await this.getOrOpen(sessionId); + async commands(ref: PiSessionLookup): Promise { + const session = await this.getOrOpen(ref); const commands: ClientCommand[] = [...BUILTIN_COMMANDS]; for (const command of session.extensionRunner.getRegisteredCommands()) { commands.push({ name: command.invocationName, ...(command.description === undefined ? {} : { description: command.description }), source: "extension" }); @@ -360,9 +377,9 @@ export class PiSessionService { return commands.sort((a, b) => a.name.localeCompare(b.name)); } - async prompt(sessionId: string, text: string, streamingBehavior?: "steer" | "followUp"): Promise { - await this.assertWritable(sessionId); - const session = await this.getOrOpen(sessionId); + async prompt(ref: PiSessionLookup, text: string, streamingBehavior?: "steer" | "followUp"): Promise { + await this.assertWritable(ref); + const session = await this.getOrOpen(ref); this.maybeGenerateSessionName(session, text); const isQueued = session.isStreaming || session.isCompacting; const behavior = isQueued ? streamingBehavior ?? "followUp" : undefined; @@ -398,9 +415,9 @@ export class PiSessionService { this.publishStatus(session); } - async shell(sessionId: string, text: string): Promise { - await this.assertWritable(sessionId); - const active = await this.getActive(sessionId); + async shell(ref: PiSessionLookup, text: string): Promise { + await this.assertWritable(ref); + const active = await this.getActive(ref); const { session } = active.runtime; const isExcluded = text.startsWith("!!"); const command = (isExcluded ? text.slice(2) : text.slice(1)).trim(); @@ -433,26 +450,28 @@ export class PiSessionService { }); } - async runCommand(sessionId: string, text: string): Promise { - await this.assertWritable(sessionId); - return this.commandService.run(sessionId, text); + async runCommand(ref: PiSessionLookup, text: string): Promise { + await this.assertWritable(ref); + const active = await this.getActive(ref); + return this.commandService.run(active.runtime.session.sessionId, text); } - async respondToCommand(sessionId: string, requestId: string, value: string): Promise { - await this.assertWritable(sessionId); - return this.commandService.respond(sessionId, requestId, value); + async respondToCommand(ref: PiSessionLookup, requestId: string, value: string): Promise { + await this.assertWritable(ref); + const active = await this.getActive(ref); + return this.commandService.respond(active.runtime.session.sessionId, requestId, value); } - async archive(sessionId: string): Promise { - const session = await this.getOrOpen(sessionId); + async archive(ref: PiSessionLookup): Promise { + const session = await this.getOrOpen(ref); if (this.hasActiveWork(session)) throw new Error("Stop current session activity before archiving"); const archiveInput = await this.archiveInputForSession(session); await this.closeActive(session.sessionId); await this.archiveStore.archive(archiveInput); } - async archiveTree(sessionId: string): Promise { - const session = await this.getOrOpen(sessionId); + async archiveTree(ref: PiSessionLookup): Promise { + const session = await this.getOrOpen(ref); const catalog = await this.workspaceArchiveCandidates(session.sessionManager.getCwd()); const root = findArchiveCandidateByIdOrPrefix(catalog, session.sessionId) ?? archiveCandidateFromActiveSession(session, false); const plan = planSessionArchiveTree(root, catalog); @@ -471,21 +490,24 @@ export class PiSessionService { }; } - async restore(sessionId: string): Promise { - await this.closeActive(sessionId); - await this.archiveStore.restore(sessionId); + async restore(ref: PiSessionLookup): Promise { + const archived = await this.getArchived(ref); + if (archived === undefined) throw new Error("Session not found"); + await this.closeActive(archived.sessionId); + await this.archiveStore.restore(archived.sessionId); } - async detachParent(sessionId: string): Promise { - const session = await this.getOrOpen(sessionId); + async detachParent(ref: PiSessionLookup): Promise { + const session = await this.getOrOpen(ref); const sessionFile = session.sessionFile; if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); await clearParentSession(sessionFile); } - async abort(sessionId: string): Promise { - const active = this.active.get(sessionId); - if (!active) return; + async abort(ref: PiSessionLookup): Promise { + const active = this.activeForLookup(ref); + if (active === undefined) return; + const sessionId = active.runtime.session.sessionId; this.clearCompactionPromptQueue(sessionId); clearSessionQueue(active.runtime.session); await active.runtime.session.abort(); @@ -493,8 +515,10 @@ export class PiSessionService { this.publishStatus(active.runtime.session); } - stop(sessionId: string): void { - void this.closeActive(sessionId).catch(() => { + stop(ref: PiSessionLookup): void { + const active = this.activeForLookup(ref); + if (active === undefined) return; + void this.closeActive(active.runtime.session.sessionId).catch(() => { // Best-effort shutdown; callers that need errors await closeActive directly. }); } @@ -591,26 +615,44 @@ export class PiSessionService { } } - private async assertWritable(sessionId: string): Promise { - if (await this.archiveStore.isArchived(sessionId)) throw new Error("Archived sessions are read-only. Restore the session to continue."); + private async assertWritable(ref: PiSessionLookup): Promise { + if (await this.getArchived(ref) !== undefined) throw new Error("Archived sessions are read-only. Restore the session to continue."); } - private async getOrOpen(sessionId: string): Promise { - return (await this.getActive(sessionId)).runtime.session; + private async getOrOpen(ref: PiSessionLookup): Promise { + return (await this.getActive(ref)).runtime.session; } - private async getActive(sessionId: string): Promise> { - const active = this.active.get(sessionId); - if (active) return active; + private async getActive(ref: PiSessionLookup): Promise> { + const active = this.activeForLookup(ref); + if (active !== undefined) return active; - const archived = await this.archiveStore.get(sessionId); + const archived = await this.getArchived(ref); if (archived?.archivePath !== undefined) return this.create(this.sessionManager.open(archived.archivePath), archived.cwd); - const match = (await this.sessionManager.listAll()).find((s) => s.id === sessionId || s.id.startsWith(sessionId)); + if (!isPiSessionRef(ref)) throw new Error("Session not found"); + const match = (await this.sessionManager.list(ref.cwd)).find((s) => s.id === ref.id || s.id.startsWith(ref.id)); if (!match) throw new Error("Session not found"); return this.create(this.sessionManager.open(match.path), match.cwd); } + private async getArchived(ref: PiSessionLookup): Promise { + const archived = await this.archiveStore.get(sessionIdFromLookup(ref)); + if (archived === undefined) return undefined; + if (isPiSessionRef(ref) && archived.cwd !== ref.cwd) return undefined; + return archived; + } + + private activeForLookup(ref: PiSessionLookup): ActiveSession | undefined { + const sessionId = sessionIdFromLookup(ref); + const exact = this.active.get(sessionId); + if (exact !== undefined && lookupMatchesActiveSession(ref, exact)) return exact; + for (const [candidateId, active] of this.active.entries()) { + if (candidateId.startsWith(sessionId) && lookupMatchesActiveSession(ref, active)) return active; + } + return undefined; + } + private async create(sessionManager: PiSessionManager, cwd: string): Promise> { const runtime = await this.createAgentRuntime(this.createRuntime, { cwd, agentDir: this.agentDir, sessionManager }); const active: ActiveSession = { runtime, unsubscribe: noop }; diff --git a/src/server/sessions/sessionRoutes.ts b/src/server/sessions/sessionRoutes.ts index 4de5b55..929648a 100644 --- a/src/server/sessions/sessionRoutes.ts +++ b/src/server/sessions/sessionRoutes.ts @@ -1,9 +1,20 @@ import type { FastifyInstance } from "fastify"; import type { SessionEventHub } from "../realtime/sessionEventHub.js"; -import type { PiSessionService } from "./piSessionService.js"; +import type { PiSessionRef, PiSessionService } from "./piSessionService.js"; + +interface SessionQuery { + cwd?: string; +} + +interface MessageQuery extends SessionQuery { + before?: string; + limit?: string; +} + +class SessionRouteValidationError extends Error {} export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionService, eventHub: SessionEventHub, prefix = ""): void { - app.get<{ Querystring: { cwd?: string } }>(`${prefix}/sessions`, async (request, reply) => { + app.get<{ Querystring: SessionQuery }>(`${prefix}/sessions`, async (request, reply) => { if (request.query.cwd === undefined || request.query.cwd === "") return reply.code(400).send({ error: "cwd query parameter is required" }); return sessions.list(request.query.cwd); }); @@ -12,164 +23,189 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS try { return await sessions.start(request.body.cwd); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.get<{ Params: { sessionId: string }; Querystring: { before?: string; limit?: string } }>(`${prefix}/sessions/:sessionId/messages`, async (request, reply) => { + 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(request.params.sessionId, page); + return await sessions.messages(sessionRefFromQuery(request.params.sessionId, request.query), page); } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); } }); - app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/status`, async (request, reply) => { + app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/status`, async (request, reply) => { try { - return await sessions.status(request.params.sessionId); + return await sessions.status(sessionRefFromQuery(request.params.sessionId, request.query)); } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); } }); - app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/models`, async (request, reply) => { + app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/models`, async (request, reply) => { try { - return { models: await sessions.availableModels(request.params.sessionId) }; + return { models: await sessions.availableModels(sessionRefFromQuery(request.params.sessionId, request.query)) }; } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string }; Body: { provider: string; modelId: string } }>(`${prefix}/sessions/:sessionId/model`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; provider?: unknown; modelId?: unknown } }>(`${prefix}/sessions/:sessionId/model`, async (request, reply) => { try { - return await sessions.setModel(request.params.sessionId, request.body.provider, request.body.modelId); + const body = requireRecord(request.body); + return await sessions.setModel(sessionRefFromBody(request.params.sessionId, body), requireString(body, "provider"), requireString(body, "modelId")); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string }; Body: { direction?: "forward" | "backward" } }>(`${prefix}/sessions/:sessionId/model/cycle`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; direction?: "forward" | "backward" } }>(`${prefix}/sessions/:sessionId/model/cycle`, async (request, reply) => { try { - return await sessions.cycleModel(request.params.sessionId, request.body.direction ?? "forward"); + const body = requireRecord(request.body); + const direction = body["direction"]; + 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"); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/thinking-levels`, async (request, reply) => { + app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/thinking-levels`, async (request, reply) => { try { - return { levels: await sessions.availableThinkingLevels(request.params.sessionId) }; + return { levels: await sessions.availableThinkingLevels(sessionRefFromQuery(request.params.sessionId, request.query)) }; } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string }; Body: { level: "off" | "minimal" | "low" | "medium" | "high" | "xhigh" } }>(`${prefix}/sessions/:sessionId/thinking-level`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; level?: unknown } }>(`${prefix}/sessions/:sessionId/thinking-level`, async (request, reply) => { try { - return await sessions.setThinkingLevel(request.params.sessionId, request.body.level); + const body = requireRecord(request.body); + return await sessions.setThinkingLevel(sessionRefFromBody(request.params.sessionId, body), requireThinkingLevel(body["level"])); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/thinking-level/cycle`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/thinking-level/cycle`, async (request, reply) => { try { - return await sessions.cycleThinkingLevel(request.params.sessionId); + const body = requireRecord(request.body); + return await sessions.cycleThinkingLevel(sessionRefFromBody(request.params.sessionId, body)); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/commands`, async (request, reply) => { + app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/commands`, async (request, reply) => { try { - return await sessions.commands(request.params.sessionId); + return await sessions.commands(sessionRefFromQuery(request.params.sessionId, request.query)); } catch (error) { - return reply.code(404).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(readErrorStatus(error)).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string }; Body: { text: string; streamingBehavior?: "steer" | "followUp" } }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown; streamingBehavior?: "steer" | "followUp" } }>(`${prefix}/sessions/:sessionId/prompt`, async (request, reply) => { try { - await sessions.prompt(request.params.sessionId, request.body.text, request.body.streamingBehavior); + const body = requireRecord(request.body); + const streamingBehavior = body["streamingBehavior"]; + if (streamingBehavior !== undefined && streamingBehavior !== "steer" && streamingBehavior !== "followUp") throw new Error("streamingBehavior must be steer or followUp"); + await sessions.prompt(sessionRefFromBody(request.params.sessionId, body), requireString(body, "text"), streamingBehavior); return { accepted: true }; } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string }; Body: { text: string } }>(`${prefix}/sessions/:sessionId/shell`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown } }>(`${prefix}/sessions/:sessionId/shell`, async (request, reply) => { try { - await sessions.shell(request.params.sessionId, request.body.text); + const body = requireRecord(request.body); + await sessions.shell(sessionRefFromBody(request.params.sessionId, body), requireString(body, "text")); return { accepted: true }; } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string }; Body: { text: string } }>(`${prefix}/sessions/:sessionId/commands/run`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; text?: unknown } }>(`${prefix}/sessions/:sessionId/commands/run`, async (request, reply) => { try { - return await sessions.runCommand(request.params.sessionId, request.body.text); + const body = requireRecord(request.body); + return await sessions.runCommand(sessionRefFromBody(request.params.sessionId, body), requireString(body, "text")); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string }; Body: { requestId: string; value: string } }>(`${prefix}/sessions/:sessionId/commands/respond`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; requestId?: unknown; value?: unknown } }>(`${prefix}/sessions/:sessionId/commands/respond`, async (request, reply) => { try { - return await sessions.respondToCommand(request.params.sessionId, request.body.requestId, request.body.value); + const body = requireRecord(request.body); + return await sessions.respondToCommand(sessionRefFromBody(request.params.sessionId, body), requireString(body, "requestId"), requireString(body, "value")); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/abort`, async (request) => { - await sessions.abort(request.params.sessionId); - return { aborted: true }; - }); - - app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/stop`, (request) => { - sessions.stop(request.params.sessionId); - return { stopped: true }; - }); - - app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/archive`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/abort`, async (request, reply) => { try { - await sessions.archive(request.params.sessionId); + await sessions.abort(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); + return { aborted: true }; + } catch (error) { + return reply.code(400).send({ error: errorMessage(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/stop`, (request, reply) => { + try { + sessions.stop(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); + return { stopped: true }; + } catch (error) { + return reply.code(400).send({ error: errorMessage(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/archive`, async (request, reply) => { + try { + await sessions.archive(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); return { archived: true }; } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/archive-tree`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/archive-tree`, async (request, reply) => { try { - return await sessions.archiveTree(request.params.sessionId); + return await sessions.archiveTree(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/restore`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/restore`, async (request, reply) => { try { - await sessions.restore(request.params.sessionId); + await sessions.restore(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); return { restored: true }; } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.post<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/detach-parent`, async (request, reply) => { + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown } }>(`${prefix}/sessions/:sessionId/detach-parent`, async (request, reply) => { try { - await sessions.detachParent(request.params.sessionId); + await sessions.detachParent(sessionRefFromBody(request.params.sessionId, requireRecord(request.body))); return { detached: true }; } catch (error) { - return reply.code(400).send({ error: error instanceof Error ? error.message : String(error) }); + return reply.code(400).send({ error: errorMessage(error) }); } }); - app.get<{ Params: { sessionId: string } }>(`${prefix}/sessions/:sessionId/events`, { websocket: true }, (socket, request) => { - eventHub.add(request.params.sessionId, socket); + app.get<{ Params: { sessionId: string }; Querystring: SessionQuery }>(`${prefix}/sessions/:sessionId/events`, { websocket: true }, (socket, request) => { + try { + const ref = sessionRefFromQuery(request.params.sessionId, request.query); + eventHub.add(ref.id, socket); + } catch { + socket.close(); + } }); app.get(`${prefix}/sessions/events`, { websocket: true }, (socket) => { @@ -181,6 +217,34 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS }); } +function sessionRefFromQuery(id: string, query: SessionQuery): PiSessionRef { + const cwd = query.cwd; + if (cwd === undefined || cwd === "") throw new SessionRouteValidationError("cwd query parameter is required"); + return { id, cwd }; +} + +function sessionRefFromBody(id: string, body: Record): PiSessionRef { + const cwd = body["cwd"]; + if (typeof cwd !== "string" || cwd === "") throw new Error("cwd field is required"); + return { id, cwd }; +} + +function requireRecord(value: unknown): Record { + if (!isRecord(value)) throw new Error("request body must be an object"); + return value; +} + +function requireString(record: Record, field: string): string { + const value = record[field]; + if (typeof value !== "string") throw new Error(`${field} field must be a string`); + return value; +} + +function requireThinkingLevel(value: unknown): "off" | "minimal" | "low" | "medium" | "high" | "xhigh" { + if (value === "off" || value === "minimal" || value === "low" || value === "medium" || value === "high" || value === "xhigh") return value; + throw new Error("level field is invalid"); +} + function optionalField(key: string, value: T | undefined): Record | object { return value === undefined ? {} : { [key]: value }; } @@ -190,3 +254,15 @@ function optionalNumber(value: string | undefined): number | undefined { const parsed = Number(value); return Number.isFinite(parsed) ? parsed : undefined; } + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function readErrorStatus(error: unknown): 400 | 404 { + return error instanceof SessionRouteValidationError ? 400 : 404; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/server/types.ts b/src/server/types.ts index 633440a..541ac8d 100644 --- a/src/server/types.ts +++ b/src/server/types.ts @@ -1,6 +1,7 @@ export type { Project, Workspace, + SessionRef as ClientSessionRef, SessionInfo as ClientSession, ArchiveSessionsResponse as ClientArchiveSessionsResponse, MessagePage as ClientMessagePage, diff --git a/src/shared/apiTypes.ts b/src/shared/apiTypes.ts index 56019c0..7d9b97f 100644 --- a/src/shared/apiTypes.ts +++ b/src/shared/apiTypes.ts @@ -86,10 +86,13 @@ export interface Workspace { isGitWorktree: boolean; } -export interface SessionInfo { +export interface SessionRef { id: string; - path: string; cwd: string; +} + +export interface SessionInfo extends SessionRef { + path: string; name?: string; created: string; modified: string;