diff --git a/src/server/sessions/piSessionService.askUser.test.ts b/src/server/sessions/piSessionService.askUser.test.ts index 99c9c09..b07f7bc 100644 --- a/src/server/sessions/piSessionService.askUser.test.ts +++ b/src/server/sessions/piSessionService.askUser.test.ts @@ -1,13 +1,20 @@ import { describe, expect, it, vi } from "vitest"; +import { ASK_USER_ANSWERS_CUSTOM_TYPE } from "../../shared/apiTypes.js"; import { createPiWebCustomToolDefinitions, PiSessionService } from "./piSessionService.js"; import { PendingAskStore, PendingAskValidationError } from "./pendingAskStore.js"; -import { CapturingSessionEventHub, emptyArchiveStore, sessionGateway, testModelRuntime } from "./piSessionService.testSupport.js"; +import { CapturingSessionEventHub, emptyArchiveStore, fakeRuntime, runtimeCreator, sessionGateway, sessionRecord, sessionRef, testModelRuntime } from "./piSessionService.testSupport.js"; const TEST_AGENT_DIR = "/tmp/pi-web-test-agent"; +const ACTIVE_SESSION_ID = "session-1"; const questions = [{ id: "db", question: "Which database?", options: [{ value: "pg", label: "Postgres" }] }]; -function askService() { +/** + * Service over a clocked store with sequential ask ids, so asks are named + * `ask-1`, `ask-2`, … and timestamps are fixed. Pass `withActiveSession` when the + * test needs a live runtime to deliver answers into. + */ +function askService(options: { withActiveSession?: boolean } = {}) { const store = new PendingAskStore({ now: () => new Date("2026-02-01T10:00:00.000Z"), createAskId: (() => { @@ -15,18 +22,28 @@ function askService() { return () => { next += 1; return `ask-${next.toString()}`; }; })(), }); - const service = new PiSessionService(new CapturingSessionEventHub(), { + const fake = fakeRuntime(ACTIVE_SESSION_ID); + const events = new CapturingSessionEventHub(); + const service = new PiSessionService(events, { agentDir: TEST_AGENT_DIR, modelRuntime: testModelRuntime, - sessionManager: sessionGateway([]), + sessionManager: sessionGateway(options.withActiveSession === true ? [sessionRecord(ACTIVE_SESSION_ID)] : []), archiveStore: emptyArchiveStore(), + ...(options.withActiveSession === true ? { createAgentRuntime: runtimeCreator(fake.runtime) } : {}), pendingAskStore: store, askUserEnabled: true, heartbeatIntervalMs: 60_000, }); - return { service, store }; + return { service, store, events, fake }; } +function askEvents(events: CapturingSessionEventHub) { + return events.sessionEvents + .filter(({ event }) => event.type === "ask.opened" || event.type === "ask.closed") + .map(({ sessionId, event }) => ({ sessionId, event })); +} + + describe("ask_user registration", () => { it("offers ask_user whenever the capability is configured, including to restricted tracked children", () => { const askUser = { open: vi.fn() }; @@ -82,11 +99,155 @@ describe("PiSessionService.openAsk", () => { }); it("rejects an unanswerable question set without opening it", async () => { - const { service, store } = askService(); + const { service, store, events } = askService(); await expect(service.openAsk({ sessionId: "session-1", questions: [{ id: "empty", question: "No way to answer?", options: [] }] })) .rejects.toThrow(PendingAskValidationError); expect(store.pendingAsk("session-1")).toBeUndefined(); + expect(askEvents(events)).toEqual([]); + await service.dispose(); + }); + + it("publishes ask.opened so a watching browser renders the card without refetching status", async () => { + const { service, events } = askService(); + + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + + expect(askEvents(events)).toEqual([ + { sessionId: ACTIVE_SESSION_ID, event: { type: "ask.opened", ask: { askId: "ask-1", askedAt: "2026-02-01T10:00:00.000Z", questions } } }, + ]); + await service.dispose(); + }); + + it("publishes the supersede as a close before the replacement opens", async () => { + const { service, events } = askService(); + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions: [{ id: "again", question: "Still?", options: [], allowOther: true }] }); + + expect(askEvents(events).map(({ event }) => event)).toEqual([ + { type: "ask.opened", ask: { askId: "ask-1", askedAt: "2026-02-01T10:00:00.000Z", questions } }, + { type: "ask.closed", askId: "ask-1", reason: "superseded" }, + { type: "ask.opened", ask: { askId: "ask-2", askedAt: "2026-02-01T10:00:00.000Z", questions: [{ id: "again", question: "Still?", options: [], allowOther: true }] } }, + ]); + await service.dispose(); + }); +}); + +describe("PiSessionService ask status projection", () => { + it("reports the open ask in status so a reloading browser rehydrates it", async () => { + const { service } = askService({ withActiveSession: true }); + + const before = await service.status(sessionRef(ACTIVE_SESSION_ID)); + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + const during = await service.status(sessionRef(ACTIVE_SESSION_ID)); + await service.submitAsk(sessionRef(ACTIVE_SESSION_ID), "ask-1", { answers: [{ id: "db", values: ["pg"] }] }); + const after = await service.status(sessionRef(ACTIVE_SESSION_ID)); + + expect(before).not.toHaveProperty("pendingAsk"); + expect(during.pendingAsk).toMatchObject({ askId: "ask-1", questions }); + expect(after).not.toHaveProperty("pendingAsk"); + await service.dispose(); + }); + + it("drops the open ask when the runtime that posted it is closed", async () => { + const { service, store } = askService({ withActiveSession: true }); + await service.status(sessionRef(ACTIVE_SESSION_ID)); + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + + await service.stop(sessionRef(ACTIVE_SESSION_ID)); + + expect(store.pendingAsk(ACTIVE_SESSION_ID)).toBeUndefined(); + await service.dispose(); + }); +}); + +describe("PiSessionService.submitAsk", () => { + it("delivers the answers as a follow-up custom message that wakes the session", async () => { + const { service, store, events, fake } = askService({ withActiveSession: true }); + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + + const response = await service.submitAsk(sessionRef(ACTIVE_SESSION_ID), "ask-1", { answers: [{ id: "db", values: ["pg"] }] }); + + expect(response).toMatchObject({ result: "closed", outcome: { askId: "ask-1", reason: "submitted", answeredCount: 1, unansweredIds: [] } }); + expect(response.sessionStatus.sessionId).toBe(ACTIVE_SESSION_ID); + expect(fake.calls.sendCustomMessage).toHaveLength(1); + const [delivered] = fake.calls.sendCustomMessage; + expect(delivered?.message.customType).toBe(ASK_USER_ANSWERS_CUSTOM_TYPE); + expect(delivered?.message.display).toBe(true); + expect(delivered?.message.content).toContain("The user submitted answers to your questions."); + expect(delivered?.message.content).toContain("Answered 1 of 1"); + expect(delivered?.message.details).toMatchObject({ askId: "ask-1", reason: "submitted" }); + expect(delivered?.options).toEqual({ triggerTurn: true, deliverAs: "followUp" }); + expect(askEvents(events).map(({ event }) => event.type)).toEqual(["ask.opened", "ask.closed"]); + expect(store.pendingAsk(ACTIVE_SESSION_ID)).toBeUndefined(); + await service.dispose(); + }); + + it("names the questions the user left unanswered in the message the model reads", async () => { + const { service, fake } = askService({ withActiveSession: true }); + await service.openAsk({ + sessionId: ACTIVE_SESSION_ID, + questions: [...questions, { id: "cache", question: "Which cache?", options: [{ value: "redis", label: "Redis" }] }], + }); + + const response = await service.submitAsk(sessionRef(ACTIVE_SESSION_ID), "ask-1", { answers: [{ id: "db", values: ["pg"] }] }); + + expect(response.outcome).toMatchObject({ answeredCount: 1, unansweredIds: ["cache"] }); + expect(fake.calls.sendCustomMessage[0]?.message.content).toContain("Answered 1 of 2; unanswered: cache"); + await service.dispose(); + }); + + it("reports a stale ask id without delivering anything or closing the open ask", async () => { + const { service, store, events, fake } = askService({ withActiveSession: true }); + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + + const response = await service.submitAsk(sessionRef(ACTIVE_SESSION_ID), "ask-superseded", { answers: [] }); + + expect(response.result).toBe("stale"); + expect(response).not.toHaveProperty("outcome"); + expect(response.sessionStatus.pendingAsk).toMatchObject({ askId: "ask-1" }); + expect(fake.calls.sendCustomMessage).toEqual([]); + expect(askEvents(events).map(({ event }) => event.type)).toEqual(["ask.opened"]); + expect(store.pendingAsk(ACTIVE_SESSION_ID)).toMatchObject({ askId: "ask-1" }); + await service.dispose(); + }); + + it("leaves the ask open when the submitted answers do not fit its questions", async () => { + const { service, store, fake } = askService({ withActiveSession: true }); + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + + await expect(service.submitAsk(sessionRef(ACTIVE_SESSION_ID), "ask-1", { answers: [{ id: "nope", values: [] }] })) + .rejects.toThrow(PendingAskValidationError); + + expect(store.pendingAsk(ACTIVE_SESSION_ID)).toMatchObject({ askId: "ask-1" }); + expect(fake.calls.sendCustomMessage).toEqual([]); + await service.dispose(); + }); +}); + +describe("PiSessionService.cancelAsk", () => { + it("tells the model every question went unanswered rather than leaving it waiting", async () => { + const { service, store, events, fake } = askService({ withActiveSession: true }); + await service.openAsk({ sessionId: ACTIVE_SESSION_ID, questions }); + + const response = await service.cancelAsk(sessionRef(ACTIVE_SESSION_ID), "ask-1"); + + expect(response).toMatchObject({ result: "closed", outcome: { reason: "cancelled", answeredCount: 0, unansweredIds: ["db"] } }); + expect(fake.calls.sendCustomMessage[0]?.message.content).toContain("closed (cancelled) before it was fully answered"); + expect(fake.calls.sendCustomMessage[0]?.options).toEqual({ triggerTurn: true, deliverAs: "followUp" }); + expect(askEvents(events).at(-1)?.event).toEqual({ type: "ask.closed", askId: "ask-1", reason: "cancelled" }); + expect(store.pendingAsk(ACTIVE_SESSION_ID)).toBeUndefined(); + await service.dispose(); + }); + + it("reports a stale cancel of an ask that is already gone", async () => { + const { service, fake } = askService({ withActiveSession: true }); + + const response = await service.cancelAsk(sessionRef(ACTIVE_SESSION_ID), "ask-1"); + + expect(response.result).toBe("stale"); + expect(fake.calls.sendCustomMessage).toEqual([]); await service.dispose(); }); }); diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index b4946b2..b9a9b72 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -33,8 +33,11 @@ import { deterministicSessionName, fallbackSessionName, generateShortSessionName import { computeEditPreview, type EditPreviewResult } from "./editPreview.js"; import { attachmentsToInlineImages, saveAttachmentsToWorkspace } from "./attachmentService.js"; import { parsePromptAttachments } from "../../shared/promptAttachments.js"; -import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_LIMIT } from "../../shared/apiTypes.js"; +import { ASK_USER_ANSWERS_CUSTOM_TYPE, SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_LIMIT } from "../../shared/apiTypes.js"; import type { + AskUserCloseResponse, + AskUserOutcome, + AskUserSubmission, SavedPromptAttachment, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, @@ -55,7 +58,7 @@ import { type AuthChange } from "./authService.js"; import { canonicalizeStoredCwd, cwdPathsEqual } from "../workingDirectory.js"; import type { WorkspaceActivityService } from "../activity/workspaceActivityService.js"; import { createAskUserToolDefinition, type AskUserInvocation, type AskUserToolDeps } from "./askUserTool.js"; -import { PendingAskStore, type PendingAskOpenResult } from "./pendingAskStore.js"; +import { PendingAskStore, renderAskUserAnswersText, type PendingAskCloseResult, type PendingAskOpenResult } from "./pendingAskStore.js"; import { createSpawnSessionToolDefinition, type SpawnSessionInvocation, type SpawnSessionResult } from "./spawnSessionTool.js"; import { createSubsessionToolDefinitions, type SpawnSubsessionInvocation, type SpawnSubsessionResult, type SubsessionCheckResult, type SubsessionReadQuery, type SubsessionReadResult, type SubsessionStatus, type SubsessionSummary, type SubsessionToolDeps } from "./spawnSubsessionTool.js"; import { buildTranscriptView } from "./subsessionTranscript.js"; @@ -952,7 +955,10 @@ export class PiSessionService implements SessionRouteService { const pendingOpens = this.pendingSessionOpenPromises(); if (pendingOpens.length > 0) await Promise.allSettled(pendingOpens); const activeSessions = Array.from(new Set(this.active.values())); - for (const active of activeSessions) this.forgetUnreadActivity(active.runtime.session); + for (const active of activeSessions) { + this.forgetUnreadActivity(active.runtime.session); + this.pendingAskStore.forgetSession(active.runtime.session.sessionId); + } this.active.clear(); this.pendingSessionOpens.clear(); this.activities.clear(); @@ -1096,7 +1102,73 @@ export class PiSessionService implements SessionRouteService { */ // eslint-disable-next-line @typescript-eslint/require-await -- async so a rejected question set becomes a rejection rather than a synchronous throw from a promise-returning method. async openAsk(input: AskUserInvocation): Promise { - return this.pendingAskStore.open(input); + const result = this.pendingAskStore.open(input); + // A supersede closes the earlier ask, so the browsers watching it must hear + // that before they hear about its replacement. + if (result.superseded !== undefined) this.publishAskClosed(input.sessionId, result.superseded); + this.events.publish(input.sessionId, { type: "ask.opened", ask: result.ask }); + this.publishStatusForSessionId(input.sessionId); + return result; + } + + /** + * Record the user's answers to the session's open ask and hand them to the + * model. The answers travel as a system-authored custom message rather than a + * user message, so they are not attributed to the human in the transcript; + * they still wake an idle session (`triggerTurn`) and queue behind in-flight + * work (`deliverAs: "followUp"`), which is how the run that `ask_user` + * terminated continues. + */ + async submitAsk(ref: PiSessionLookup, askId: string, submission: AskUserSubmission): Promise { + await this.assertWritable(ref); + const session = await this.getOrOpen(ref); + // Checked before the store closes the ask so a refused delivery cannot + // discard answers the user already submitted. + this.assertTreeNavigationInactive(session, "answer questions"); + return this.closeAsk(session, this.pendingAskStore.submit(session.sessionId, askId, submission)); + } + + /** + * Close the open ask without answers. The model is still told, naming every + * question as unanswered: it was promised a follow-up message and would + * otherwise wait for one that never comes. + */ + async cancelAsk(ref: PiSessionLookup, askId: string): Promise { + await this.assertWritable(ref); + const session = await this.getOrOpen(ref); + this.assertTreeNavigationInactive(session, "dismiss questions"); + return this.closeAsk(session, this.pendingAskStore.cancel(session.sessionId, askId)); + } + + /** + * Publish and deliver a closed ask. A stale close is reported rather than + * thrown: losing the race against a supersede, another browser, or a session + * that went away is ordinary, and the returned status tells the browser what + * the session's open ask is now. + */ + private async closeAsk(session: PiAgentSession, result: PendingAskCloseResult): Promise { + if (result.status === "stale") return { result: "stale", sessionStatus: this.statusFromSession(session) }; + const { outcome } = result; + this.publishAskClosed(session.sessionId, outcome); + await this.runSessionEntryMutation(session, "deliver answers to your questions", () => session.sendCustomMessage( + { customType: ASK_USER_ANSWERS_CUSTOM_TYPE, content: renderAskUserAnswersText(outcome), display: true, details: outcome }, + { triggerTurn: true, deliverAs: "followUp" }, + )); + this.publishStatus(session); + return { result: "closed", outcome, sessionStatus: this.statusFromSession(session) }; + } + + private publishAskClosed(sessionId: string, outcome: AskUserOutcome): void { + this.events.publish(sessionId, { type: "ask.closed", askId: outcome.askId, reason: outcome.reason }); + } + + /** + * Publish status for a session known only by id, as the ask tools are: they + * run inside the session's own runtime, so the active entry is the session. + */ + private publishStatusForSessionId(sessionId: string): void { + const session = this.active.get(sessionId)?.runtime.session; + if (session !== undefined) this.publishStatus(session); } /** Summaries of the tracked subsessions spawned by `parentSessionId`. */ @@ -2251,6 +2323,9 @@ export class PiSessionService implements SessionRouteService { } if (!active) return; this.forgetUnreadActivity(active.runtime.session); + // An open ask is meaningful only while the runtime that posted it exists: no + // one is left to receive the answers, so it is dropped without an outcome. + this.pendingAskStore.forgetSession(sessionId); this.active.delete(sessionId); this.activities.delete(sessionId); this.workspaceActivity?.removeSession(sessionId, active.runtime.session.sessionManager.getCwd()); @@ -3107,6 +3182,7 @@ export class PiSessionService implements SessionRouteService { const model = session.model === undefined ? undefined : modelToClientModel(session.model); const contextUsage = session.getContextUsage(); const warnings = this.warningsForSession(session); + const pendingAsk = this.pendingAskStore.pendingAsk(session.sessionId); return { sessionId: session.sessionId, persisted: sessionFileExists(session.sessionFile), @@ -3122,6 +3198,7 @@ export class PiSessionService implements SessionRouteService { cost: stats.cost, ...(contextUsage === undefined ? {} : { contextUsage }), ...(warnings.length === 0 ? {} : { warnings }), + ...(pendingAsk === undefined ? {} : { pendingAsk }), }; } diff --git a/src/server/sessions/sessionRoutes.test.ts b/src/server/sessions/sessionRoutes.test.ts index caddfa9..aea0dd2 100644 --- a/src/server/sessions/sessionRoutes.test.ts +++ b/src/server/sessions/sessionRoutes.test.ts @@ -2,8 +2,10 @@ import { resolve } from "node:path"; import Fastify, { type FastifyInstance } from "fastify"; import fastifyWebsocket from "@fastify/websocket"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_CATALOG_ID_MAX_LENGTH } from "../../shared/apiTypes.js"; +import { ASK_USER_ID_MAX_LENGTH, ASK_USER_OTHER_TEXT_MAX_LENGTH, ASK_USER_QUESTION_LIMIT, SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_CATALOG_ID_MAX_LENGTH } from "../../shared/apiTypes.js"; import type { + AskUserCloseResponse, + AskUserSubmission, MessagePage, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, @@ -337,6 +339,100 @@ describe("session routes", () => { } }); + it("parses ask submissions and reports both closed and stale outcomes", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const submitted = await routeApp.inject({ + method: "POST", + url: "/sessions/session-1/ask/submit", + payload: { + cwd: "/repo/./", + askId: "ask-1", + answers: [{ id: "db", values: ["pg"] }, { id: "cache", values: [], otherText: "redis" }], + }, + }); + const cancelled = await routeApp.inject({ + method: "POST", + url: "/sessions/session-1/ask/cancel", + payload: { askId: "ask-2" }, + }); + + expect(submitted.statusCode).toBe(200); + expect(submitted.json()).toMatchObject({ result: "closed", sessionStatus: { sessionId: "session-1" } }); + expect(routeService.submitAskCalls).toEqual([{ + lookup: { id: "session-1", cwd: resolve("/repo") }, + askId: "ask-1", + submission: { answers: [{ id: "db", values: ["pg"] }, { id: "cache", values: [], otherText: "redis" }] }, + }]); + expect(cancelled.statusCode).toBe(200); + expect(cancelled.json()).toMatchObject({ result: "stale" }); + expect(routeService.cancelAskCalls).toEqual([{ lookup: "session-1", askId: "ask-2" }]); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + + it("rejects malformed ask payloads before calling the service", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(); + registerSessionRoutes(routeApp, routeService, eventHub); + const malformed: Record[] = [ + { answers: [] }, + { askId: "", answers: [] }, + { askId: "x".repeat(ASK_USER_ID_MAX_LENGTH + 1), answers: [] }, + { askId: "ask-1" }, + { askId: "ask-1", answers: {} }, + { askId: "ask-1", answers: [{ values: ["pg"] }] }, + { askId: "ask-1", answers: [{ id: "db" }] }, + { askId: "ask-1", answers: [{ id: "db", values: [1] }] }, + { askId: "ask-1", answers: [{ id: "db", values: [], otherText: 7 }] }, + { askId: "ask-1", answers: [{ id: "db", values: [], otherText: "x".repeat(ASK_USER_OTHER_TEXT_MAX_LENGTH + 1) }] }, + { askId: "ask-1", answers: new Array(ASK_USER_QUESTION_LIMIT + 1).fill({ id: "db", values: [] }) }, + ]; + + try { + for (const payload of malformed) { + const response = await routeApp.inject({ method: "POST", url: "/sessions/session-1/ask/submit", payload }); + expect(response.statusCode).toBe(400); + } + const cancelWithoutAskId = await routeApp.inject({ method: "POST", url: "/sessions/session-1/ask/cancel", payload: {} }); + + expect(cancelWithoutAskId.statusCode).toBe(400); + expect(routeService.submitAskCalls).toEqual([]); + expect(routeService.cancelAskCalls).toEqual([]); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + + it("maps a missing session on an ask submission to 404", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(); + routeService.askError = new Error("Session not found"); + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const response = await routeApp.inject({ method: "POST", url: "/sessions/session-1/ask/submit", payload: { askId: "ask-1", answers: [] } }); + + expect(response.statusCode).toBe(404); + expect(response.json()).toEqual({ error: "Session not found" }); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + it("rejects prompt payloads that omit text without opening a session", async () => { const response = await app.inject({ method: "POST", url: "/sessions/session-1/prompt", payload: { body: "Build the thing" } }); @@ -706,9 +802,24 @@ class CapturingRouteSessionService implements SessionRouteService { readonly bulkArchiveCalls: SessionBulkMutationRef[][] = []; readonly bulkDeleteCalls: SessionBulkMutationRef[][] = []; readonly navigateTreeCalls: { lookup: SessionRouteLookup; request: SessionTreeNavigateRequest }[] = []; + readonly submitAskCalls: { lookup: SessionRouteLookup; askId: string; submission: AskUserSubmission }[] = []; + readonly cancelAskCalls: { lookup: SessionRouteLookup; askId: string }[] = []; + askError: Error | undefined; reloadError: Error | undefined; clearQueueError: Error | undefined; + submitAsk(lookup: SessionRouteLookup, askId: string, submission: AskUserSubmission): Promise { + if (this.askError !== undefined) return Promise.reject(this.askError); + this.submitAskCalls.push({ lookup, askId, submission }); + return Promise.resolve({ result: "closed", sessionStatus: idleStatus(lookup) }); + } + + cancelAsk(lookup: SessionRouteLookup, askId: string): Promise { + if (this.askError !== undefined) return Promise.reject(this.askError); + this.cancelAskCalls.push({ lookup, askId }); + return Promise.resolve({ result: "stale", sessionStatus: idleStatus(lookup) }); + } + cleanupPreview(request: NormalizedSessionCleanupRequest): Promise { this.cleanupPreviewCalls.push(request); return Promise.resolve({ generatedAt: "2026-06-25T00:00:00.000Z", thresholds: request.thresholds, projects: [], totals: { archiveCount: 0, deleteCount: 0 } }); @@ -901,6 +1012,19 @@ function notificationSnapshot(ref: SessionRef): SessionNotificationInboxSnapshot }; } +function idleStatus(lookup: SessionRouteLookup): SessionStatus { + return { + sessionId: sessionIdFromLookup(lookup), + isStreaming: false, + isCompacting: false, + isBashRunning: false, + pendingMessageCount: 0, + queuedMessages: [], + tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + cost: 0, + }; +} + function sessionIdFromLookup(lookup: SessionRouteLookup): string { return typeof lookup === "string" ? lookup : lookup.id; } diff --git a/src/server/sessions/sessionRoutes.ts b/src/server/sessions/sessionRoutes.ts index da73425..b528539 100644 --- a/src/server/sessions/sessionRoutes.ts +++ b/src/server/sessions/sessionRoutes.ts @@ -1,5 +1,5 @@ import type { FastifyInstance } from "fastify"; -import { SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_CATALOG_ID_MAX_LENGTH, SESSION_UNREAD_CWD_MAX_LENGTH, SESSION_UNREAD_SESSION_ID_MAX_LENGTH, type SessionBulkMutationRequest, type SessionBulkMutationRef, type SessionCleanupRequest, type SessionTreeNavigateRequest, type SessionTreeSummaryChoice, type SessionUnreadAcknowledgeRequest } from "../../shared/apiTypes.js"; +import { ASK_USER_ID_MAX_LENGTH, ASK_USER_OPTION_LIMIT, ASK_USER_OTHER_TEXT_MAX_LENGTH, ASK_USER_QUESTION_LIMIT, SESSION_TREE_CUSTOM_INSTRUCTIONS_MAX_LENGTH, SESSION_UNREAD_CATALOG_ID_MAX_LENGTH, SESSION_UNREAD_CWD_MAX_LENGTH, SESSION_UNREAD_SESSION_ID_MAX_LENGTH, type AskUserAnswer, type AskUserSubmission, type SessionBulkMutationRequest, type SessionBulkMutationRef, type SessionCleanupRequest, type SessionTreeNavigateRequest, type SessionTreeSummaryChoice, type SessionUnreadAcknowledgeRequest } from "../../shared/apiTypes.js"; import { projectBrowserMessageResponse } from "../browserMessageProjection.js"; import { normalizeRequestCwd } from "../workingDirectory.js"; import type { SessionEventHub } from "../realtime/sessionEventHub.js"; @@ -266,6 +266,25 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: SessionRou } }); + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; askId?: unknown; answers?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/ask/submit`, async (request, reply) => { + try { + const body = requireRecord(request.body); + const askId = requireBoundedId(body["askId"], "askId"); + return await sessions.submitAsk(sessionLookupFromBody(request.params.sessionId, body), askId, askUserSubmissionFromBody(body)); + } catch (error) { + return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) }); + } + }); + + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; askId?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/ask/cancel`, async (request, reply) => { + try { + const body = requireRecord(request.body); + return await sessions.cancelAsk(sessionLookupFromBody(request.params.sessionId, body), requireBoundedId(body["askId"], "askId")); + } catch (error) { + return reply.code(mutationErrorStatus(error)).send({ error: errorMessage(error) }); + } + }); + app.post<{ Params: { sessionId: string }; Body: { cwd?: unknown; dismissId?: unknown } | undefined }>(`${prefix}/sessions/:sessionId/warnings/dismiss`, async (request, reply) => { try { const body = optionalRecord(request.body); @@ -493,6 +512,37 @@ function requireExactFields(record: Record, fields: readonly st if (unexpected !== undefined) throw new Error(`${label} field contains unsupported property: ${unexpected}`); } +/** + * Shape-check one submitted answer set. Only transport-level checks belong here: + * whether the answers fit the questions that were actually asked is the pending + * ask store's job, since only it knows the open ask. + */ +function askUserSubmissionFromBody(body: Record): AskUserSubmission { + const answers = body["answers"]; + if (!Array.isArray(answers)) throw new Error("answers field must be an array"); + if (answers.length > ASK_USER_QUESTION_LIMIT) throw new Error("answers field has too many entries"); + return { answers: answers.map(askUserAnswerFromValue) }; +} + +function askUserAnswerFromValue(value: unknown): AskUserAnswer { + const record = requireRecord(value); + const values = record["values"]; + if (!Array.isArray(values)) throw new Error("values field must be an array"); + if (values.length > ASK_USER_OPTION_LIMIT) throw new Error("values field has too many entries"); + const otherText = record["otherText"]; + if (otherText !== undefined && typeof otherText !== "string") throw new Error("otherText field must be a string"); + if (typeof otherText === "string" && otherText.length > ASK_USER_OTHER_TEXT_MAX_LENGTH) throw new Error("otherText field is too long"); + return { + id: requireBoundedId(record["id"], "id"), + values: values.map((entry) => requireBoundedId(entry, "values entry")), + ...(otherText === undefined ? {} : { otherText }), + }; +} + +function requireBoundedId(value: unknown, field: string): string { + return requireNonEmptyBoundedString(value, field, ASK_USER_ID_MAX_LENGTH); +} + function optionalRecord(value: unknown): Record { if (value === undefined || value === null) return {}; return requireRecord(value); diff --git a/src/server/sessions/sessionService.ts b/src/server/sessions/sessionService.ts index 7e64414..d7ef88e 100644 --- a/src/server/sessions/sessionService.ts +++ b/src/server/sessions/sessionService.ts @@ -1,4 +1,6 @@ import type { + AskUserCloseResponse, + AskUserSubmission, SavedPromptAttachment, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, @@ -51,6 +53,8 @@ export interface SessionRouteService { dismissNotification(ref: SessionRouteRef, request: Omit): SessionNotificationInboxSnapshot | Promise; dismissAllNotifications(ref: SessionRouteRef, request: Omit): SessionNotificationInboxSnapshot | Promise; clearQueue(ref: SessionRouteLookup): Promise; + submitAsk(ref: SessionRouteLookup, askId: string, submission: AskUserSubmission): Promise; + cancelAsk(ref: SessionRouteLookup, askId: string): Promise; dismissWarning(ref: SessionRouteLookup, dismissId: string): Promise; availableModels(ref: SessionRouteLookup): Promise; setModel(ref: SessionRouteLookup, provider: string, modelId: string): Promise; diff --git a/src/shared/apiTypes.ts b/src/shared/apiTypes.ts index de24563..a4fff79 100644 --- a/src/shared/apiTypes.ts +++ b/src/shared/apiTypes.ts @@ -440,6 +440,12 @@ export interface QueuedSessionMessage { text: string; } +/** + * `customType` of the follow-up custom message that carries a closed ask back to + * the model and into the transcript. Its `details` are an {@link AskUserOutcome}. + */ +export const ASK_USER_ANSWERS_CUSTOM_TYPE = "pi-web.ask.answers"; + /** Largest question set one `ask_user` call may post. */ export const ASK_USER_QUESTION_LIMIT = 20; /** Largest option list one question may offer. */ @@ -544,6 +550,21 @@ export interface AskUserOutcome { summary: string; } +/** + * Result of the browser closing an ask by submitting or cancelling it. + * + * `"stale"` is an ordinary race rather than an error: the named ask was already + * submitted, superseded by a newer one, or gone with its session runtime. The + * browser drops its card and trusts `sessionStatus`, which is returned in both + * cases so closing an ask needs no follow-up status request. + */ +export interface AskUserCloseResponse { + result: "closed" | "stale"; + /** Present only when this call is the one that closed the ask. */ + outcome?: AskUserOutcome; + sessionStatus: SessionStatus; +} + /** * Progress of the session startup window, where the daemon is still * constructing the agent session and no `PiAgentSession` exists yet, so diff --git a/src/shared/federatedRoutes.ts b/src/shared/federatedRoutes.ts index 80f5447..32e4de2 100644 --- a/src/shared/federatedRoutes.ts +++ b/src/shared/federatedRoutes.ts @@ -68,6 +68,8 @@ export const FEDERATED_HTTP_ROUTES = [ { method: "GET", path: "/sessions/:sessionId/commands" }, { method: "POST", path: "/sessions/:sessionId/prompt" }, { method: "POST", path: "/sessions/:sessionId/queue/clear" }, + { method: "POST", path: "/sessions/:sessionId/ask/submit" }, + { method: "POST", path: "/sessions/:sessionId/ask/cancel" }, { method: "POST", path: "/sessions/:sessionId/warnings/dismiss" }, { method: "POST", path: "/sessions/:sessionId/attachments" }, { method: "POST", path: "/sessions/:sessionId/shell" },