diff --git a/docs/architecture.md b/docs/architecture.md index fdf7ec909a..4ed25ba1c4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -345,6 +345,10 @@ join active work or reuse committed stages. A persisted resource alone cannot pr that a provider accepted its initial prompt. Interrupted side effects with no conclusive receipt return an unknown outcome instead of being repeated. +Agent registration commits creation. Rejecting the first prompt is a turn error on that +agent; reporting creation failure leaves an orphaned agent and invites another Create. +Receipt replay must not resend the prompt. + `packages/client/src/creation/` owns capability selection and legacy orchestration. Callers always pass the initial prompt to agent creation. On an older host, the client adapts keyed creation to the legacy create/send sequence; it cannot continue that diff --git a/packages/app/e2e/browser/agent-message-submission.spec.ts b/packages/app/e2e/browser/agent-message-submission.spec.ts index 75cf7ff0c9..3bfda6b4b0 100644 --- a/packages/app/e2e/browser/agent-message-submission.spec.ts +++ b/packages/app/e2e/browser/agent-message-submission.spec.ts @@ -284,7 +284,6 @@ async function expectRejectedSubmissionRestored( await expectComposerDraft(page, input.prompt); await expectComposerEditable(page); await expectAttachmentPill(page, "composer-image-attachment-pill"); - await expect(page.getByTestId("user-message").filter({ hasText: input.prompt })).toHaveCount(0); if (input.preservesActiveTurn) { await expect(page.getByTestId("turn-working-indicator")).toBeVisible(); return; @@ -295,13 +294,21 @@ async function expectRejectedSubmissionRestored( async function retryRestoredSubmission(page: Page, prompt: string): Promise { await page.getByRole("textbox", { name: "Message agent..." }).first().press("Enter"); - const userMessage = page.getByTestId("user-message").filter({ hasText: prompt }); - await expect(userMessage).toHaveCount(1); + const attempts = page.getByTestId("user-message").filter({ hasText: prompt }); + await expect(attempts).toHaveCount(2); + const userMessage = attempts.last(); await expect(userMessage).toHaveAttribute("aria-busy", "false", { timeout: 30_000 }); await expect(userMessage.getByRole("button", { name: "Open image attachment" })).toBeVisible(); await expect(page.getByTestId("composer-image-attachment-pill")).toHaveCount(0); } +async function expectRejectedPromptBeforeError(page: Page, prompt: string, errorMessage: string) { + const messages = page.getByTestId("user-message").or(page.getByTestId("assistant-message")); + await expect(messages).toHaveCount(2); + await expect(messages.nth(0)).toContainText(prompt); + await expect(messages.nth(1)).toContainText(`[System Error] ${errorMessage}`); +} + async function configureSteerInSettings(page: Page): Promise { const modifier = process.platform === "darwin" ? "Meta" : "Control"; await page.keyboard.press(`${modifier}+Comma`); @@ -1173,13 +1180,14 @@ test.describe("Agent message submission", () => { await toasts.expectNeverShown("agent-updating-toast"); }); - test("restores a rejected submission and accepts its retry", async ({ + test("retains a rejected submission before its error and accepts its retry", async ({ page, rejectionScenario, }) => { const prompt = "Restore this rejected submission."; await submitMessageThatWillBeRejected(page, prompt); await expectRejectedSubmissionRestored(page, { prompt, ...rejectionScenario }); + await expectRejectedPromptBeforeError(page, prompt, rejectionScenario.errorMessage); await retryRestoredSubmission(page, prompt); }); @@ -1209,6 +1217,7 @@ test.describe("Agent message submission", () => { errorMessage: "Requested mock steer transport failure", preservesActiveTurn: true, }); + await expect(page.getByTestId("user-message").filter({ hasText: prompt })).toHaveCount(0); expect(gate.getClientRequestCount("send_agent_message_request")).toBe(sendsBefore + 1); expect(gate.getClientRequestCount("cancel_agent_request")).toBe(cancelsBefore); diff --git a/packages/app/e2e/browser/creation-idempotency.spec.ts b/packages/app/e2e/browser/creation-idempotency.spec.ts index 39e73dbded..7658addf72 100644 --- a/packages/app/e2e/browser/creation-idempotency.spec.ts +++ b/packages/app/e2e/browser/creation-idempotency.spec.ts @@ -128,3 +128,46 @@ for (const scenario of [ await creation.expectAgentCount(1); }); } + +for (const entry of [ + { + name: "agent draft", + open: (creation: Awaited>) => + creation.openAgentDraft(), + button: "Send message", + settled: async (_creation: Awaited>) => {}, + }, + { + name: "new workspace", + open: (creation: Awaited>) => + creation.openWorkspaceForm("local"), + button: "Create", + settled: (creation: Awaited>) => + creation.expectOneCreatedWorkspace(), + }, +]) { + test(`${entry.name} becomes an agent tab when the first prompt is rejected`, async ({ + creation, + promptRejection, + page, + }, testInfo) => { + void promptRejection; + const prompt = + "Continue the attached conversation.\n" + "Earlier conversation context.\n".repeat(100); + await entry.open(creation); + await creation.submitPrompt(prompt, entry.button); + await entry.settled(creation); + await creation.expectCreatedAgentError(); + await creation.expectPromptBeforeError(prompt); + await page.screenshot({ path: testInfo.outputPath("created-agent-prompt-error.png") }); + await creation.reloadAgent(); + await creation.expectPromptBeforeError(prompt); + await creation.submitPrompt("emit 1 coalesced agent stream updates for a corrected prompt."); + await creation.expectPromptVisible( + "emit 1 coalesced agent stream updates for a corrected prompt.", + ); + await creation.expectAssistantReply(); + await creation.expectAgentCount(1); + await expect(page.getByText("agent_request_key_conflict", { exact: true })).toHaveCount(0); + }); +} diff --git a/packages/app/e2e/support/creation-fixtures.ts b/packages/app/e2e/support/creation-fixtures.ts index 33308ee386..9608a95a0c 100644 --- a/packages/app/e2e/support/creation-fixtures.ts +++ b/packages/app/e2e/support/creation-fixtures.ts @@ -13,6 +13,7 @@ import { delayBrowserAgentCreatedStatus } from "./helpers/new-workspace"; export const test = base.extend<{ startup: { release(): Promise; fail(): Promise }; + promptRejection: void; creation: Awaited>; promptRetry: ReturnType; agentRetries: Awaited>; @@ -65,6 +66,37 @@ export default function contribute(server) { await client.close(); } }, + promptRejection: async ({ page }, provide) => { + void page; + const client = await connectNewWorkspaceDaemonClient(); + const plugin = await copyPluginExample("lifecycle-logger"); + const previous = await client.getDaemonConfig(); + try { + await writeFile( + path.join(plugin.directory, "index.server.ts"), + ` +export default function contribute(server) { + return server.before("agent.create", ({ request }) => ({ + ...request, + config: { + ...request.config, + model: "e2e-fast-stream", + featureValues: { ...request.config.featureValues, mockPromptRejections: 1 }, + }, + })); +} +`, + ); + await client.patchDaemonConfig({ pluginsEnabled: true }); + await client.installDirectoryPlugin(plugin.directory); + await provide(); + } finally { + await client.removePlugin("lifecycle-logger"); + await client.patchDaemonConfig({ pluginsEnabled: previous.config.pluginsEnabled }); + await plugin.cleanup(); + await client.close(); + } + }, creation: async ({ page }, provide) => { const creation = await createCreationScenario(page); try { diff --git a/packages/app/e2e/support/helpers/creation.ts b/packages/app/e2e/support/helpers/creation.ts index 09cd2ea290..9098068698 100644 --- a/packages/app/e2e/support/helpers/creation.ts +++ b/packages/app/e2e/support/helpers/creation.ts @@ -182,6 +182,47 @@ export async function createCreationScenario(page: Page) { await this.expectWorkspaceReadyBeforeAgentCompletion(); expect((await project.client.fetchAgents()).entries).toHaveLength(0); }, + async expectCreatedAgentError() { + await requests.settled(); + await expect( + page + .getByTestId("assistant-message") + .filter({ hasText: "[System Error] Requested mock prompt rejection", visible: true }), + ).toBeVisible(); + await expect( + page + .locator('[data-testid^="workspace-tab-agent_"][aria-selected="true"]') + .filter({ visible: true }), + ).toBeVisible(); + await expect( + page.locator('[data-testid^="workspace-tab-draft_"]').filter({ visible: true }), + ).toHaveCount(0); + await expect(page.getByRole("textbox", { name: "Message agent..." }).first()).toBeEditable(); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + }, + async expectPromptBeforeError(prompt: string) { + const messages = page + .getByTestId("user-message") + .or(page.getByTestId("assistant-message")) + .filter({ visible: true }); + await expect(messages).toHaveCount(2); + await expect(messages.nth(0)).toContainText(prompt); + await expect(messages.nth(1)).toContainText("[System Error]"); + await expect(messages.nth(1)).toBeInViewport(); + }, + async reloadAgent() { + await page.reload(); + await expect(page.getByText("Updating messages", { exact: true })).toHaveCount(0); + await this.expectCreatedAgentError(); + }, + async expectAssistantReply() { + await expect( + page + .getByTestId("assistant-message") + .filter({ visible: true, hasNotText: "[System Error]" }) + .first(), + ).toBeVisible(); + }, async expectStartupFailure() { await expect(page.getByText(/Creation startup failed for test/).first()).toBeVisible(); }, diff --git a/packages/server/package.json b/packages/server/package.json index 64125950db..5201d4cae4 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -89,7 +89,7 @@ "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts", "test": "npm run test:unit && npm run test:integration", "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"", - "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\" && vitest run --maxWorkers=1 src/server/agent/providers/opencode-bridge.local.e2e.test.ts -t \"versioned runtime\"", + "test:integration": "vitest run --maxWorkers=1 src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\" && vitest run --maxWorkers=1 src/server/agent/providers/opencode-bridge.local.e2e.test.ts -t \"versioned runtime\"", "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts", "test:integration:all": "npm run test:e2e", "test:integration:real": "vitest run real.e2e.test.ts", diff --git a/packages/server/src/server/agent/activity-curator.test.ts b/packages/server/src/server/agent/activity-curator.test.ts index 15ad9db90e..68f864174a 100644 --- a/packages/server/src/server/agent/activity-curator.test.ts +++ b/packages/server/src/server/agent/activity-curator.test.ts @@ -438,3 +438,49 @@ second line'`, ).toThrow("Selected assistant message is no longer available."); }); }); + +it("bounds fork history to 100,000 characters while preserving the newest messages", () => { + const result = buildAgentForkContextAttachment({ + rows: [ + row(1, { type: "user_message", text: "old-message " + "x".repeat(1_100_000) }), + row(2, { type: "assistant_message", messageId: "checkpoint", text: "Latest answer." }), + row(3, { type: "user_message", text: "Beyond the selected checkpoint." }), + ], + boundaryMessageId: "checkpoint", + }); + expect(result.attachment.text.length).toBeLessThanOrEqual(100_000); + expect(result.attachment.text).toContain("[Earlier chat history omitted]"); + expect(result.attachment.text).toContain("[Assistant] Latest answer."); + expect(result.attachment.text).not.toContain("old-message"); + expect(result.attachment.text).not.toContain("Beyond the selected checkpoint."); + expect(result.attachment.text).toMatch(/^\n/); + expect(result.attachment.text).toMatch(/\n<\/chat-history-summary>$/); +}); + +it("bounds one oversized fork message and metadata without losing its beginning and end", () => { + const result = buildAgentForkContextAttachment({ + rows: [ + row(1, { + type: "assistant_message", + text: "Beginning. " + "x".repeat(1_100_000) + " The end.", + }), + ], + agentTitle: "T".repeat(200_000), + cwd: "/" + "p".repeat(200_000), + }); + expect(result.attachment.text.length).toBe(100_000); + expect(result.attachment.text).toContain("[Assistant] Beginning."); + expect(result.attachment.text).toContain("[Message truncated]"); + expect(result.attachment.text).toContain("The end."); + expect(result.attachment.text).toMatch(/\n<\/chat-history-summary>$/); +}); + +it("preserves every message in fork history below the character budget", () => { + const text = "Complete conversation " + "x".repeat(90_000); + const result = buildAgentForkContextAttachment({ + rows: [row(1, { type: "user_message", text })], + }); + expect(result.attachment.text).toBe( + `\nChat history from a previous Paseo agent.\n\n[User] ${text}\n`, + ); +}); diff --git a/packages/server/src/server/agent/activity-curator.ts b/packages/server/src/server/agent/activity-curator.ts index 3bcab550e7..b8381c9b7c 100644 --- a/packages/server/src/server/agent/activity-curator.ts +++ b/packages/server/src/server/agent/activity-curator.ts @@ -8,6 +8,9 @@ import { projectTimelineRows } from "./timeline-projection.js"; const DEFAULT_MAX_ITEMS = 0; const MAX_TOOL_INPUT_CHARS = 400; const MAX_TOOL_SUMMARY_CHARS = 200; +const MAX_FORK_CONTEXT_CHARS = 100_000; +const MAX_FORK_METADATA_CHARS = 1_000; +const FORK_HISTORY_OMITTED = "[Earlier chat history omitted]\n"; interface ActivityCuratorOptions { maxItems?: number; @@ -276,7 +279,7 @@ function selectForkContextRows(input: { function trimContextMetadata(value: string | null | undefined): string | null { const trimmed = value?.trim(); - return trimmed ? trimmed : null; + return trimmed ? trimmed.slice(0, MAX_FORK_METADATA_CHARS) : null; } function buildForkContextText(input: { @@ -296,6 +299,30 @@ function buildForkContextText(input: { return `\n${header.join("\n")}\n\n${input.body}\n`; } +function fitForkContextEntries(entries: ActivityEntry[], budget: number): string { + const length = entries.reduce((total, entry) => total + entry.text.length + 1, -1); + if (length <= budget) return entries.map((entry) => entry.text).join("\n"); + + const selected: string[] = []; + let remaining = budget - FORK_HISTORY_OMITTED.length; + for (let index = entries.length - 1; index >= 0; index--) { + const text = entries[index]!.text; + const size = text.length + (selected.length > 0 ? 1 : 0); + if (size > remaining) { + if (selected.length === 0) { + const marker = "\n[Message truncated]\n"; + const available = remaining - marker.length; + const head = Math.ceil(available / 2); + selected.push(text.slice(0, head) + marker + text.slice(-(available - head))); + } + break; + } + selected.push(text); + remaining -= size; + } + return FORK_HISTORY_OMITTED + selected.toReversed().join("\n"); +} + export function buildAgentForkContextAttachment(input: { rows: readonly AgentTimelineRow[]; cursorBoundary?: ForkCursorBoundary | null; @@ -319,10 +346,10 @@ export function buildAgentForkContextAttachment(input: { includeKinds: ["user_message", "assistant_message", "tool_call"], includeExternalToolInput: false, }); + const metadata = { agentTitle: input.agentTitle, cwd: input.cwd }; + const budget = MAX_FORK_CONTEXT_CHARS - buildForkContextText({ body: "", ...metadata }).length; const body = - entries.length > 0 - ? entries.map((entry) => entry.text).join("\n") - : "No chat history to display."; + entries.length > 0 ? fitForkContextEntries(entries, budget) : "No chat history to display."; return { attachment: { type: "text", diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 629159d51f..cd8bd2004a 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -3954,12 +3954,23 @@ test("a prompt after provider replacement reopens the stale session", async () = clients: { [provider]: replacement }, }); - const dispatch = await startAgentRun(manager, created.id, "continue after reload", logger); + const dispatch = await startAgentRun(manager, created.id, "continue after reload", logger, { + runOptions: { clientMessageId: "stale-session-retry" }, + }); expect(dispatch.disposition).toBe("turn_started"); await replacement.waitForRetryStart(); const result = await manager.waitForAgentEvent(created.id); expect(result.status).toBe("idle"); expect(replacement.resumeOverrides).toHaveLength(1); + const timeline = manager.getTimeline(created.id); + expect(timeline.filter((item) => item.type === "user_message")).toMatchObject([ + { text: "continue after reload", clientMessageId: "stale-session-retry" }, + ]); + expect( + timeline.some( + (item) => item.type === "assistant_message" && item.text.includes("[System Error]"), + ), + ).toBe(false); } finally { await manager.closeAgent(created.id).catch(() => undefined); await manager.flush().catch(() => undefined); @@ -8141,9 +8152,20 @@ test("streamAgent clears pending run when startTurn fails before a turn id exist { workspaceId: undefined }, ); - await expect(manager.runAgent(agent.id, "fail before turn id")).rejects.toThrow( - "Invalid request: missing field `text`", - ); + await expect( + manager.runAgent(agent.id, "fail before turn id", { + clientMessageId: "rejected-prompt", + }), + ).rejects.toThrow("Invalid request: missing field `text`"); + expect(manager.getTimeline(agent.id)).toEqual([ + { + type: "user_message", + text: "fail before turn id", + clientMessageId: "rejected-prompt", + messageId: "rejected-prompt", + }, + { type: "assistant_message", text: "[System Error] Invalid request: missing field `text`" }, + ]); await expect(manager.runAgent(agent.id, "second turn")).resolves.toEqual( expect.objectContaining({ diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 390717cf66..734e4ab8c7 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2360,7 +2360,7 @@ export class AgentManager { return false; } if (options?.clientMessageId) { - this.recordSubmittedPrompt(agent, prompt, options.clientMessageId); + this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "accepted"); this.emitState(agent); } const dispatch = (event: AgentStreamEvent): void => { @@ -2456,6 +2456,12 @@ export class AgentManager { agent.pendingReplacement = false; const errorMsg = error instanceof Error ? error.message : "Failed to start turn"; pendingRun.start = { status: "failed", error: errorMsg }; + // A terminal rejection belongs after the submitted prompt even though no provider turn exists. + if (options?.clientMessageId) { + this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "rejected", { + messageId: options.clientMessageId, + }); + } await this.handleStreamEvent(agent, { type: "turn_failed", provider: agent.provider, @@ -2545,7 +2551,7 @@ export class AgentManager { ) : undefined; if (options?.clientMessageId) { - this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, { + this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "accepted", { messageId: options.clientMessageId, turnId, providerMessageId: @@ -2851,7 +2857,7 @@ export class AgentManager { if (!clientMessageId) { return; } - this.recordSubmittedPrompt(agent, prompt, clientMessageId, { + this.recordSubmittedPrompt(agent, prompt, clientMessageId, "accepted", { messageId: clientMessageId, turnId: expectedTurnId, }); @@ -4696,6 +4702,7 @@ export class AgentManager { agent: ActiveManagedAgent, prompt: AgentPromptInput, clientMessageId: string, + outcome: "accepted" | "rejected", options?: { messageId?: string; providerMessageId?: string; turnId?: string }, ): void { const item = projectAgentMessage({ @@ -4704,7 +4711,10 @@ export class AgentManager { clientMessageId, ...(options?.messageId ? { messageId: options.messageId } : {}), }); - if (item) this.recordSubmittedPromptItem(agent, item, options); + if (!item) return; + // Human attempts stay in history on rejection; delivery notifications require acceptance. + if (outcome === "rejected" && item.type !== "user_message") return; + this.recordSubmittedPromptItem(agent, item, options); } private recordSubmittedPromptItem( diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index eaafad96b6..ab1ce218b1 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -300,7 +300,8 @@ async function resolveSessionCreateAgent( runOptions, setupContinuation, background: true, - promptFailure: "throw", + // Registration commits creation. First-turn failures belong to the created agent. + promptFailure: "return-error", promptLogger: dependencies.logger.child({ clientMessageId: resolveClientMessageId(input.clientMessageId), }), @@ -481,7 +482,11 @@ async function sendInitialPrompt( throw error; } if (resolved.promptFailure === "return-error") { - return { started: false, liveSnapshot: snapshot, error }; + return { + started: false, + liveSnapshot: dependencies.agentManager.getAgent(snapshot.id) ?? snapshot, + error, + }; } dependencies.logger.error({ err: error, agentId: snapshot.id }, "Failed to run initial prompt"); return { started: false, liveSnapshot: snapshot }; diff --git a/packages/server/src/server/creation/creation.e2e.test.ts b/packages/server/src/server/creation/creation.e2e.test.ts index f0d21be18f..f4d35bd3f3 100644 --- a/packages/server/src/server/creation/creation.e2e.test.ts +++ b/packages/server/src/server/creation/creation.e2e.test.ts @@ -1,5 +1,6 @@ import { execFileSync } from "node:child_process"; import type { z } from "zod"; +import { BUILTIN_PROVIDER_IDS } from "@getpaseo/protocol/provider-manifest"; import { WebSocket } from "ws"; import { SessionInboundMessageSchema, WSOutboundMessageSchema } from "@getpaseo/protocol/messages"; import { randomUUID } from "node:crypto"; @@ -9,10 +10,23 @@ import { join } from "node:path"; import { expect, test, vi } from "vitest"; import pino from "pino"; import type { CreationSnapshot, SessionOutboundMessage } from "@getpaseo/protocol/messages"; +import type { AgentTimelineItem } from "../agent/agent-sdk-types.js"; import { DaemonClient } from "../test-utils/daemon-client.js"; import { createTestAgentClients } from "../test-utils/fake-agent-client.js"; import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +function createCreationDaemon(options: Parameters[0] = {}) { + const agentClients = options.agentClients ?? createTestAgentClients(); + return createTestPaseoDaemon({ + ...options, + agentClients, + // Installed provider CLIs must not change when background naming starts. + providerOverrides: Object.fromEntries( + BUILTIN_PROVIDER_IDS.map((id) => [id, { enabled: id in agentClients }]), + ), + }); +} + function deferred() { let resolve!: (value: T) => void; const promise = new Promise((done) => { @@ -29,7 +43,7 @@ test("creation progresses before agent readiness and continues after the disconn const directory = await mkdtemp(join(tmpdir(), "creation-wire-")); let agents = 0; let prompts = 0; - const daemon = await createTestPaseoDaemon({ + const daemon = await createCreationDaemon({ logger: pino( { level: "trace" }, { @@ -41,11 +55,13 @@ test("creation progresses before agent readiness and continues after the disconn }, ), agentClients: createTestAgentClients({ - beforeCreateSession: async () => { + beforeCreateSession: async (config) => { + if (config.internal) return; agents++; await provider.promise; }, - onStartTurn: () => { + onStartTurn: (_prompt, config) => { + if (config.internal) return; prompts++; }, }), @@ -122,11 +138,12 @@ test("creation progresses before agent readiness and continues after the disconn await client.close(); await observer.close(); await daemon.close(); - await rm(directory, { recursive: true, force: true }); + await rm(directory, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 }); } }, 60000); async function connectCreationPeer(port: number) { + const timeout = 30_000; const socket = new WebSocket(`ws://127.0.0.1:${port}/ws`); const frames: SessionOutboundMessage[] = []; socket.on("message", (data) => { @@ -147,7 +164,9 @@ async function connectCreationPeer(port: number) { }), ); await expect - .poll(() => frames.some((m) => m.type === "status" && m.payload.status === "server_info")) + .poll(() => frames.some((m) => m.type === "status" && m.payload.status === "server_info"), { + timeout, + }) .toBe(true); return { close: () => socket.close(), @@ -160,7 +179,7 @@ async function connectCreationPeer(port: number) { (m) => "payload" in m && "requestId" in m.payload && m.payload.requestId === message.requestId, ); - await expect.poll(response).toBeDefined(); + await expect.poll(response, { timeout }).toBeDefined(); return response()!; }, }; @@ -170,7 +189,7 @@ test.each([false, true])( "workspace identity does not depend on subscribing (first subscribe=%s)", async (subscribe) => { const directory = await mkdtemp(join(tmpdir(), "creation-identity-")); - const daemon = await createTestPaseoDaemon(); + const daemon = await createCreationDaemon(); const peer = await connectCreationPeer(daemon.port); try { const request = { @@ -188,7 +207,7 @@ test.each([false, true])( } finally { peer.close(); await daemon.close(); - await rm(directory, { recursive: true, force: true }); + await rm(directory, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 }); } }, 60000, @@ -199,10 +218,13 @@ test.each(["create_agent_request", "agent.create.request"] as const)( async (type) => { const directory = await mkdtemp(join(tmpdir(), "creation-agent-identity-")); let creations = 0; - const daemon = await createTestPaseoDaemon({ + const daemon = await createCreationDaemon({ agentClients: createTestAgentClients({ - beforeCreateSession: async () => { + beforeCreateSession: async (config) => { + if (config.internal) return; creations++; + // A real provider or Git checkout may take longer than Vitest's default 1s poll budget. + await new Promise((resolve) => setTimeout(resolve, 1_200)); }, }), }); @@ -241,7 +263,7 @@ test.each(["create_agent_request", "agent.create.request"] as const)( } finally { peer.close(); await daemon.close(); - await rm(directory, { recursive: true, force: true }); + await rm(directory, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 }); } }, 60000, @@ -264,7 +286,7 @@ test("legacy keyed creation preserves checkout error codes", async () => { ], { cwd: directory, stdio: "pipe" }, ); - const daemon = await createTestPaseoDaemon({ agentClients: createTestAgentClients() }); + const daemon = await createCreationDaemon({ agentClients: createTestAgentClients() }); const peer = await connectCreationPeer(daemon.port); try { const result = await peer.request({ @@ -281,6 +303,95 @@ test("legacy keyed creation preserves checkout error codes", async () => { } finally { peer.close(); await daemon.close(); - await rm(directory, { recursive: true, force: true }); + await rm(directory, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 }); } }, 60000); + +test.each(["agent", "workspace"] as const)( + "%s creation succeeds when the provider rejects its initial prompt", + async (kind) => { + const directory = await mkdtemp(join(tmpdir(), "creation-prompt-rejected-")); + const rejection = "Input exceeds the maximum length of 1048576 characters."; + let sessions = 0; + let prompts = 0; + const daemon = await createCreationDaemon({ + agentClients: createTestAgentClients({ + beforeCreateSession: async (config) => { + if (config.internal) return; + sessions++; + }, + onStartTurn: (_prompt, config) => { + if (config.internal) return; + prompts++; + throw new Error(rejection); + }, + }), + }); + const client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.11.1", + }); + const snapshots: CreationSnapshot[] = []; + const agent = { + config: { provider: "codex", cwd: directory }, + initialPrompt: "Continue the attached conversation.", + clientMessageId: "draft:initial-message", + }; + const streamed: AgentTimelineItem[] = []; + const unsubscribe = daemon.daemon.agentManager.subscribe((event) => { + if (event.type === "agent_stream" && event.event.type === "timeline") { + streamed.push(event.event.item); + } + }); + try { + await client.connect(); + const submit = async () => { + const common = { + idempotencyKey: "rejected-first-prompt", + onEvent: (snapshot: CreationSnapshot) => snapshots.push(snapshot), + }; + if (kind === "agent") return client.createAgent({ ...common, ...agent }); + const result = await client.createWorkspace({ + ...common, + source: { kind: "directory", path: directory }, + agent, + }); + expect(result.error).toBeNull(); + return result.agent!; + }; + const created = await submit(); + expect(created).toMatchObject({ status: "error", lastError: rejection }); + expect(snapshots.map((snapshot) => snapshot.phase)).not.toContain("prompt_started"); + expect(snapshots.at(-1)).toMatchObject({ + phase: "completed", + error: null, + agent: { id: created.id }, + }); + const expectedTimeline = [ + { + type: "user_message", + text: agent.initialPrompt, + clientMessageId: agent.clientMessageId, + messageId: agent.clientMessageId, + }, + { type: "assistant_message", text: `[System Error] ${rejection}` }, + ]; + expect( + (await client.fetchAgentTimeline(created.id)).entries.map((entry) => entry.item), + ).toEqual(expectedTimeline); + expect(streamed).toEqual(expectedTimeline); + expect((await submit()).id).toBe(created.id); + expect( + (await client.fetchAgentTimeline(created.id)).entries.map((entry) => entry.item), + ).toEqual(expectedTimeline); + expect((await client.fetchAgents()).entries).toHaveLength(1); + expect({ sessions, prompts }).toEqual({ sessions: 1, prompts: 1 }); + } finally { + unsubscribe(); + await client.close(); + await daemon.close(); + await rm(directory, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 }); + } + }, + 60000, +); diff --git a/packages/server/src/server/creation/index.test.ts b/packages/server/src/server/creation/index.test.ts index 1eaf4a7fd9..a165110ef8 100644 --- a/packages/server/src/server/creation/index.test.ts +++ b/packages/server/src/server/creation/index.test.ts @@ -93,7 +93,7 @@ async function fixture() { calls.push("agent"); await onReady(agent); calls.push("prompt"); - return agent; + return { agent, initialPromptStarted: true }; }, }; const updates: CreationSnapshot[] = []; @@ -129,7 +129,7 @@ test("a failed agent startup retries only that stage with the reserved IDs", asy expect(readyWorkspace).toEqual(workspace); if (attempts === 1) throw new Error("provider unavailable"); await onReady(agent); - return agent; + return { agent, initialPromptStarted: true }; }; expect(await f.service.create(f.input)).toMatchObject({ phase: "failed", @@ -210,7 +210,7 @@ test("omitted resource IDs are generated once and preserved on replay", async () f.input.createAgent = async (id, createdWorkspace, onReady) => { const created = { ...agent, id, workspaceId: createdWorkspace!.id }; await onReady(created); - return created; + return { agent: created, initialPromptStarted: true }; }; const result = await f.service.create(f.input); expect(result.workspaceId).toMatch(/^wks_[a-f0-9]{16}$/); @@ -231,7 +231,7 @@ test("restart after a committed agent without a prompt completes without recreat await onReady(agent); // Capture the durable state at the exact milestone a process could stop after. await cp(f.directory, recoveredDirectory, { recursive: true }); - return agent; + return { agent, initialPromptStarted: false }; }; await f.service.create(f.input); const recovered = await new CreationService(recoveredDirectory, silentLogger).create(f.input); diff --git a/packages/server/src/server/creation/index.ts b/packages/server/src/server/creation/index.ts index 3d7a42efdb..e0b28b8069 100644 --- a/packages/server/src/server/creation/index.ts +++ b/packages/server/src/server/creation/index.ts @@ -13,6 +13,10 @@ import { writeFileAtomic } from "../atomic-file.js"; import { generateWorkspaceId } from "../workspace-registry-model.js"; type Observer = (snapshot: CreationSnapshot) => void; +export interface CreatedAgent { + agent: AgentSnapshotPayload; + initialPromptStarted: boolean; +} interface CreationRequest { key: string; request: { [key: string]: unknown }; @@ -28,7 +32,7 @@ interface CreationRequest { agentId: string, workspace: WorkspaceDescriptorPayload | undefined, onReady: (agent: AgentSnapshotPayload) => Promise, - ) => Promise; + ) => Promise; } export type CreationInput = CreationRequest & ( @@ -167,7 +171,7 @@ export class CreationService { if (input.hasAgent && input.createAgent && !record.snapshot.agent) { record.inFlight = "agent"; await this.write(identity, record); - const agent = await input.createAgent( + const { agent, initialPromptStarted } = await input.createAgent( record.snapshot.agentId!, record.snapshot.workspace, async (readyAgent) => { @@ -180,7 +184,7 @@ export class CreationService { }, ); record.inFlight = null; - if (input.hasPrompt) + if (initialPromptStarted) await this.publish(identity, record, { phase: "prompt_started", agent }); else record.snapshot = { ...record.snapshot, agent }; } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index f837276dec..7ac679c35a 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -8,7 +8,7 @@ import type { } from "@getpaseo/protocol/messages"; import { relative } from "node:path"; import { isAbsolute } from "node:path"; -import { CreationService } from "./creation/index.js"; +import { CreationService, type CreatedAgent } from "./creation/index.js"; import type { CreationSnapshot, AgentCreateRequest } from "@getpaseo/protocol/messages"; import type { MessageReceipts } from "./message-receipts/index.js"; import equal from "fast-deep-equal"; @@ -4173,7 +4173,7 @@ export class Session { if (!record) throw new Error("Previously created agent no longer exists"); agent = this.buildStoredAgentPayload(record); } else { - agent = await this.createSessionAgent(msg); + ({ agent } = await this.createSessionAgent(msg)); } this.emit({ type: "status", @@ -4212,7 +4212,7 @@ export class Session { msg: CreateAgentRequestMessage, agentId?: string, onAgentReady?: (agent: AgentSnapshotPayload) => Promise, - ): Promise { + ): Promise { const { config, worktreeName, @@ -4270,7 +4270,7 @@ export class Session { throw new Error(`Working directory does not exist or is not a directory: ${resolvedCwd}`); } - const { snapshot, liveSnapshot } = await createAgentCommand( + const { snapshot, liveSnapshot, initialPromptStarted } = await createAgentCommand( { agentManager: this.agentManager, agentStorage: this.agentStorage, @@ -4327,7 +4327,7 @@ export class Session { { agentId: snapshot.id, provider: snapshot.provider }, "Created agent", ); - return this.buildAgentPayload(liveSnapshot); + return { agent: await this.buildAgentPayload(liveSnapshot), initialPromptStarted }; } catch (error) { await this.createAgentLifecycleDispatch.cleanupCreatedWorktreeAfterFailedAgentCreate({ createdWorktree: createdWorktreeForCleanup, diff --git a/packages/server/src/server/test-utils/fake-agent-client.ts b/packages/server/src/server/test-utils/fake-agent-client.ts index 2a437f9bbf..659dd89686 100644 --- a/packages/server/src/server/test-utils/fake-agent-client.ts +++ b/packages/server/src/server/test-utils/fake-agent-client.ts @@ -58,13 +58,13 @@ interface FakeAgentSessionOptions { sessionId?: string; memoryMarker?: string | null; closeSession?: () => Promise; - onStartTurn?: (prompt: AgentPromptInput) => void; + onStartTurn?: (prompt: AgentPromptInput, config: AgentSessionConfig) => void; } export interface TestAgentClientOptions { - beforeCreateSession?: () => Promise; + beforeCreateSession?: (config: AgentSessionConfig) => Promise; closeSession?: () => Promise; - onStartTurn?: (prompt: AgentPromptInput) => void; + onStartTurn?: (prompt: AgentPromptInput, config: AgentSessionConfig) => void; supportsMcpServers?: boolean; } @@ -337,7 +337,7 @@ class FakeAgentSession implements AgentSession { private activeForegroundTurnId: string | null = null; private readonly closeSession: (() => Promise) | undefined; - private readonly onStartTurn: ((prompt: AgentPromptInput) => void) | undefined; + private readonly onStartTurn: TestAgentClientOptions["onStartTurn"]; constructor(options: FakeAgentSessionOptions) { this.capabilities = { @@ -440,7 +440,7 @@ class FakeAgentSession implements AgentSession { const turnId = `fake-turn-${this.nextTurnOrdinal++}`; this.activeForegroundTurnId = turnId; - this.onStartTurn?.(prompt); + this.onStartTurn?.(prompt, this.config); void this.emitTurnEvents(prompt); @@ -1213,7 +1213,7 @@ class FakeAgentClient implements AgentClient { config: AgentSessionConfig, _launchContext?: AgentLaunchContext, ): Promise { - await this.options.beforeCreateSession?.(); + await this.options.beforeCreateSession?.(config); return new FakeAgentSession({ providerName: this.provider, config: { ...config },