diff --git a/packages/agent/src/server/agent-server.test.ts b/packages/agent/src/server/agent-server.test.ts index 087f2b55b7..bce20573d8 100644 --- a/packages/agent/src/server/agent-server.test.ts +++ b/packages/agent/src/server/agent-server.test.ts @@ -2267,6 +2267,106 @@ describe("AgentServer HTTP Mode", () => { expect(prompt).toHaveBeenCalledTimes(4); }, 20000); + it("steers an active turn without emitting a separate turn completion", async () => { + const s = createServer(); + await s.start(); + const prompt = vi.fn(async () => ({ + stopReason: "end_turn", + _meta: { steer: true }, + })); + const broadcastTurnComplete = vi.fn(); + const resetTurnMessages = vi.fn(); + const serverInternals = s as unknown as { + activeOwnedTurnCount: number; + broadcastTurnComplete: typeof broadcastTurnComplete; + session: { + clientConnection: { prompt: typeof prompt }; + logWriter: { resetTurnMessages: typeof resetTurnMessages }; + }; + }; + serverInternals.activeOwnedTurnCount = 1; + serverInternals.broadcastTurnComplete = broadcastTurnComplete; + serverInternals.session.clientConnection.prompt = prompt; + serverInternals.session.logWriter.resetTurnMessages = resetTurnMessages; + + const response = await fetch(`http://localhost:${port}/command`, { + method: "POST", + headers: { + Authorization: `Bearer ${createToken()}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: "steer-1", + method: "user_message", + params: { content: "change direction", steer: true }, + }), + }); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toMatchObject({ + result: { stopReason: "steered", steered: true }, + }); + expect(prompt).toHaveBeenCalledWith( + expect.objectContaining({ + _meta: expect.objectContaining({ steer: true }), + }), + ); + expect(broadcastTurnComplete).not.toHaveBeenCalled(); + expect(resetTurnMessages).not.toHaveBeenCalled(); + }, 20000); + + it("declines steering without blocking on a fallback normal turn", async () => { + const s = createServer(); + await s.start(); + const prompt = vi.fn(); + const broadcastTurnComplete = vi.fn(); + const resetTurnMessages = vi.fn(); + const serverInternals = s as unknown as { + activeOwnedTurnCount: number; + broadcastTurnComplete: typeof broadcastTurnComplete; + session: { + clientConnection: { prompt: typeof prompt }; + logWriter: { resetTurnMessages: typeof resetTurnMessages }; + }; + }; + serverInternals.activeOwnedTurnCount = 1; + prompt.mockImplementationOnce(async () => { + serverInternals.activeOwnedTurnCount = 0; + return { stopReason: "end_turn", _meta: { steer: false } }; + }); + serverInternals.broadcastTurnComplete = broadcastTurnComplete; + serverInternals.session.clientConnection.prompt = prompt; + serverInternals.session.logWriter.resetTurnMessages = resetTurnMessages; + + const response = await fetch(`http://localhost:${port}/command`, { + method: "POST", + headers: { + Authorization: `Bearer ${createToken()}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: "steer-race", + method: "user_message", + params: { content: "continue normally", steer: true }, + }), + }); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toMatchObject({ + result: { stopReason: "steer_declined", steered: false }, + }); + expect(prompt).toHaveBeenCalledTimes(1); + expect(prompt.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ + _meta: expect.objectContaining({ steer: true }), + }), + ); + expect(resetTurnMessages).not.toHaveBeenCalled(); + expect(broadcastTurnComplete).not.toHaveBeenCalled(); + }, 20000); + it("redelivers a messageId whose first delivery failed before producing a turn", async () => { const s = createServer(); await s.start(); @@ -2307,6 +2407,163 @@ describe("AgentServer HTTP Mode", () => { expect(prompt).toHaveBeenCalledTimes(2); }, 20000); + it("keeps a recoverable delivery committed across an ambiguous retry", async () => { + const s = createServer(); + await s.start(); + const prompt = vi + .fn() + .mockRejectedValue(new Error("API Error: The operation timed out.")); + const serverInternals = s as unknown as { + session: { clientConnection: { prompt: typeof prompt } }; + }; + serverInternals.session.clientConnection.prompt = prompt; + + const token = createToken(); + const send = async (requestId: string) => + fetch(`http://localhost:${port}/command`, { + method: "POST", + headers: { + Authorization: `Bearer ${token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: requestId, + method: "user_message", + params: { + content: "do the thing", + messageId: "m-recoverable", + }, + }), + }); + + const first = await send("first-attempt"); + await expect(first.json()).resolves.toMatchObject({ + result: { stopReason: "error_recoverable" }, + }); + expect(prompt).toHaveBeenCalledTimes(1); + + const retry = await send("ambiguous-retry"); + await expect(retry.json()).resolves.toMatchObject({ + result: { stopReason: "duplicate_delivery", duplicate: true }, + }); + expect(prompt).toHaveBeenCalledTimes(1); + }, 20000); + + it("shares a failed in-flight messageId outcome with concurrent retries", async () => { + const s = createServer(); + await s.start(); + let rejectFirstDelivery!: (error: Error) => void; + const prompt = vi + .fn() + .mockImplementationOnce( + () => + new Promise((_resolve, reject) => { + rejectFirstDelivery = reject; + }), + ) + .mockResolvedValueOnce({ stopReason: "end_turn" }); + const serverInternals = s as unknown as { + logger: { info: (...args: unknown[]) => void }; + session: { clientConnection: { prompt: typeof prompt } }; + }; + serverInternals.session.clientConnection.prompt = prompt; + const infoLog = vi.spyOn(serverInternals.logger, "info"); + + const token = createToken(); + const send = async (requestId: string) => + fetch(`http://localhost:${port}/command`, { + method: "POST", + headers: { + Authorization: `Bearer ${token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: requestId, + method: "user_message", + params: { content: "do the thing", messageId: "m-concurrent" }, + }), + }); + + const firstResponse = send("first-attempt"); + await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(1)); + + let retrySettled = false; + const retryResponse = send("concurrent-retry").finally(() => { + retrySettled = true; + }); + await vi.waitFor(() => { + expect(infoLog).toHaveBeenCalledWith( + "Awaiting in-flight user_message delivery", + { messageId: "m-concurrent" }, + ); + expect(prompt).toHaveBeenCalledTimes(1); + expect(retrySettled).toBe(false); + }); + + rejectFirstDelivery(new Error("sdk connection lost")); + const [first, retry] = await Promise.all([firstResponse, retryResponse]); + await expect(first.json()).resolves.toMatchObject({ + error: { message: "sdk connection lost" }, + }); + await expect(retry.json()).resolves.toMatchObject({ + error: { message: "sdk connection lost" }, + }); + expect(prompt).toHaveBeenCalledTimes(1); + }, 20000); + + it("keeps an accepted messageId committed when teardown clears the active session", async () => { + const s = createServer(); + await s.start(); + let finishPrompt!: (result: { stopReason: "end_turn" }) => void; + const prompt = vi.fn( + () => + new Promise<{ stopReason: "end_turn" }>((resolve) => { + finishPrompt = resolve; + }), + ); + const serverInternals = s as unknown as { + session: { clientConnection: { prompt: typeof prompt } } | null; + }; + const acceptedSession = serverInternals.session; + if (!acceptedSession) throw new Error("expected active test session"); + acceptedSession.clientConnection.prompt = prompt; + + const token = createToken(); + const send = async (requestId: string) => + fetch(`http://localhost:${port}/command`, { + method: "POST", + headers: { + Authorization: `Bearer ${token}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: requestId, + method: "user_message", + params: { content: "do the thing", messageId: "m-teardown" }, + }), + }); + + const firstResponse = send("first-attempt"); + await vi.waitFor(() => expect(prompt).toHaveBeenCalledTimes(1)); + + serverInternals.session = null; + finishPrompt({ stopReason: "end_turn" }); + const first = await firstResponse; + await expect(first.json()).resolves.toMatchObject({ + result: { stopReason: "end_turn" }, + }); + + serverInternals.session = acceptedSession; + const retry = await send("retry"); + await expect(retry.json()).resolves.toMatchObject({ + result: { stopReason: "duplicate_delivery", duplicate: true }, + }); + expect(prompt).toHaveBeenCalledTimes(1); + }, 20000); + // Shared plumbing for the relay-echo tests: install a controllable // prompt, stub the log writer so relayAgentResponse has an answer to // relay, and spy on the relay_message client call. @@ -2447,6 +2704,7 @@ describe("AgentServer HTTP Mode", () => { expect(runStarted?.notification?.params).toMatchObject({ runId: "test-run-id", taskId: "test-task-id", + steering: "native", }); // Agent reports its semver so clients can gate UI features // against agent capabilities (e.g. `>=0.40.1`). The exact value diff --git a/packages/agent/src/server/agent-server.ts b/packages/agent/src/server/agent-server.ts index 208c643c4b..c25336e948 100644 --- a/packages/agent/src/server/agent-server.ts +++ b/packages/agent/src/server/agent-server.ts @@ -298,6 +298,15 @@ function isManualCompactPrompt(prompt: ContentBlock[]): boolean { return /^\/compact(?:\s|$)/.test(promptBlocksToText(prompt).trimStart()); } +function extractSteeringCapability(result: unknown): string | undefined { + const steering = ( + result as { + agentCapabilities?: { _meta?: { posthog?: { steering?: unknown } } }; + } + )?.agentCapabilities?._meta?.posthog?.steering; + return typeof steering === "string" ? steering : undefined; +} + interface LocalSkillPromptContext { /** Set when the message is a bare `/skill` invocation the adapter should strip. */ skillName?: string; @@ -377,6 +386,8 @@ export class AgentServer { private preSessionEvents: Record[] = []; private deliveredMessageIds = new Set(); private pendingCompactContinuationMessageIds = new Set(); + private inFlightMessageDeliveries = new Map>(); + private activeOwnedTurnCount = 0; private pendingPermissions = new Map< string, { @@ -937,49 +948,51 @@ export class AgentServer { switch (method) { case POSTHOG_NOTIFICATIONS.USER_MESSAGE: case "user_message": { - this.logger.debug("Received user_message command", { - hasContent: - typeof params.content === "string" - ? params.content.trim().length > 0 - : Array.isArray(params.content) && params.content.length > 0, - artifactCount: Array.isArray(params.artifacts) - ? params.artifacts.length - : 0, - }); - const builtPrompt = await this.buildPromptFromContentAndArtifacts({ - content: params.content as string | ContentBlock[] | undefined, - artifacts: Array.isArray(params.artifacts) - ? (params.artifacts as TaskRunArtifact[]) - : [], - taskId: this.session.payload.task_id, - runId: this.session.payload.run_id, - }); - const prompt = builtPrompt.prompt; - if (prompt.length === 0) { - throw new Error("User message cannot be empty"); - } - + const commandSession = this.session; const messageId = typeof params.messageId === "string" && params.messageId ? params.messageId : undefined; + const inFlightDelivery = messageId + ? this.inFlightMessageDeliveries.get(messageId) + : undefined; + if (inFlightDelivery) { + this.logger.info("Awaiting in-flight user_message delivery", { + messageId, + }); + return await inFlightDelivery; + } + let retryCompactContinuation = false; - if (messageId) { - if (this.deliveredMessageIds.has(messageId)) { - if (this.pendingCompactContinuationMessageIds.has(messageId)) { - retryCompactContinuation = true; - this.logger.info("Retrying pending compact continuation", { - messageId, - }); - } else { - this.logger.info("Duplicate user_message delivery ignored", { - messageId, - }); - return { stopReason: "duplicate_delivery", duplicate: true }; - } + if (messageId && this.deliveredMessageIds.has(messageId)) { + if (this.pendingCompactContinuationMessageIds.has(messageId)) { + retryCompactContinuation = true; + this.logger.info("Retrying pending compact continuation", { + messageId, + }); } else { - this.deliveredMessageIds.add(messageId); + this.logger.info("Duplicate user_message delivery ignored", { + messageId, + }); + return { stopReason: "duplicate_delivery", duplicate: true }; } + } + + let resolveDelivery: (result: unknown) => void = () => {}; + let rejectDelivery: (error: unknown) => void = () => {}; + const deliveryOutcome = new Promise((resolve, reject) => { + resolveDelivery = resolve; + rejectDelivery = reject; + }); + void deliveryOutcome.catch(() => {}); + if (messageId) { + this.inFlightMessageDeliveries.set(messageId, deliveryOutcome); + } + let deliveryCommitted = retryCompactContinuation; + const commitDelivery = (): void => { + deliveryCommitted = true; + if (!messageId) return; + this.deliveredMessageIds.add(messageId); if (this.deliveredMessageIds.size > 500) { const oldest = this.deliveredMessageIds.values().next().value; if (oldest !== undefined) { @@ -987,137 +1000,213 @@ export class AgentServer { this.pendingCompactContinuationMessageIds.delete(oldest); } } - } - this.logger.debug("Built user_message prompt", { - blockTypes: prompt.map((block) => block.type), - }); - const promptPreview = promptBlocksToText(prompt); + }; - this.logger.debug( - `Processing user message (detectedPrUrl=${this.detectedPrUrl ?? "none"}): ${promptPreview.substring(0, 100)}...`, - ); + try { + this.logger.debug("Received user_message command", { + hasContent: + typeof params.content === "string" + ? params.content.trim().length > 0 + : Array.isArray(params.content) && params.content.length > 0, + artifactCount: Array.isArray(params.artifacts) + ? params.artifacts.length + : 0, + }); + const builtPrompt = await this.buildPromptFromContentAndArtifacts({ + content: params.content as string | ContentBlock[] | undefined, + artifacts: Array.isArray(params.artifacts) + ? (params.artifacts as TaskRunArtifact[]) + : [], + taskId: commandSession.payload.task_id, + runId: commandSession.payload.run_id, + }); + const prompt = builtPrompt.prompt; + if (prompt.length === 0) { + throw new Error("User message cannot be empty"); + } - this.session.logWriter.resetTurnMessages(this.session.payload.run_id); + this.logger.debug("Built user_message prompt", { + blockTypes: prompt.map((block) => block.type), + }); + const promptPreview = promptBlocksToText(prompt); - // Resolve before buildDetectedPrContext so a warm auto-publish upgrade - // also flips the detected-PR context to its push variant. - const autoPublishUpgrade = await this.resolveWarmAutoPublishUpgrade(); - const hostContext = [ - ...(autoPublishUpgrade ? [autoPublishUpgrade] : []), - ...(this.detectedPrUrl - ? [this.buildDetectedPrContext(this.detectedPrUrl)] - : []), - ]; - const promptMeta: Record = { - ...(builtPrompt.meta ?? {}), - ...(hostContext.length > 0 - ? { prContext: hostContext.join("\n\n") } - : {}), - }; + this.logger.debug( + `Processing user message (detectedPrUrl=${this.detectedPrUrl ?? "none"}): ${promptPreview.substring(0, 100)}...`, + ); - const manualCompactPrompt = isManualCompactPrompt(prompt); - const acpSessionId = this.session.acpSessionId; - const continueAfterCompaction = (): Promise => - this.promptWithUpstreamRetry({ - sessionId: acpSessionId, - prompt: [ - hiddenTextBlock( - "Compaction is complete. Continue working on the task from the compacted context, following the user's instructions from the /compact command.", - ), - ], - }); + // Resolve before buildDetectedPrContext so a warm auto-publish upgrade + // also flips the detected-PR context to its push variant. + const autoPublishUpgrade = await this.resolveWarmAutoPublishUpgrade(); + const hostContext = [ + ...(autoPublishUpgrade ? [autoPublishUpgrade] : []), + ...(this.detectedPrUrl + ? [this.buildDetectedPrContext(this.detectedPrUrl)] + : []), + ]; + const promptMeta: Record = { + ...(builtPrompt.meta ?? {}), + ...(hostContext.length > 0 + ? { prContext: hostContext.join("\n\n") } + : {}), + }; - let compactCommandCompleted = retryCompactContinuation; - let result: PromptResponse; - this.suppressAdapterTurnComplete = - manualCompactPrompt || retryCompactContinuation; - try { - if (retryCompactContinuation) { - result = await continueAfterCompaction(); - if (messageId) { - this.pendingCompactContinuationMessageIds.delete(messageId); + if (params.steer === true) { + if (this.activeOwnedTurnCount > 0) { + const result = await commandSession.clientConnection.prompt({ + sessionId: commandSession.acpSessionId, + prompt, + _meta: { ...promptMeta, steer: true }, + }); + const accepted = + (result._meta as { steer?: unknown } | undefined)?.steer === + true; + if (accepted) { + commitDelivery(); + const outcome = { stopReason: "steered", steered: true }; + resolveDelivery(outcome); + return outcome; + } } - } else { - result = await this.session.clientConnection.prompt({ - sessionId: this.session.acpSessionId, - prompt, - ...(Object.keys(promptMeta).length > 0 - ? { _meta: promptMeta } - : {}), + const outcome = { + stopReason: "steer_declined", + steered: false, + }; + resolveDelivery(outcome); + return outcome; + } + + commandSession.logWriter.resetTurnMessages( + commandSession.payload.run_id, + ); + + const manualCompactPrompt = isManualCompactPrompt(prompt); + const acpSessionId = commandSession.acpSessionId; + const continueAfterCompaction = (): Promise => + this.promptWithUpstreamRetry({ + sessionId: acpSessionId, + prompt: [ + hiddenTextBlock( + "Compaction is complete. Continue working on the task from the compacted context, following the user's instructions from the /compact command.", + ), + ], }); - if (result.stopReason === "end_turn" && manualCompactPrompt) { - compactCommandCompleted = true; - if (messageId) { - this.pendingCompactContinuationMessageIds.add(messageId); - } - // `/compact` is an SDK-local command, so without a follow-up the - // cloud run reports completion before the model resumes the task. - this.recordTurnUsage(result.usage); - result = await continueAfterCompaction(); + let result: PromptResponse; + this.suppressAdapterTurnComplete = + manualCompactPrompt || retryCompactContinuation; + try { + if (retryCompactContinuation) { + result = await this.runOwnedTurn(continueAfterCompaction); if (messageId) { this.pendingCompactContinuationMessageIds.delete(messageId); } + } else { + result = await this.runOwnedTurn(() => { + const promptResult = commandSession.clientConnection.prompt({ + sessionId: commandSession.acpSessionId, + prompt, + ...(Object.keys(promptMeta).length > 0 + ? { _meta: promptMeta } + : {}), + }); + if (!promptResult) { + throw new Error("Agent connection did not accept the prompt"); + } + return promptResult; + }); + + if (result.stopReason === "end_turn" && manualCompactPrompt) { + commitDelivery(); + if (messageId) { + this.pendingCompactContinuationMessageIds.add(messageId); + } + // `/compact` is an SDK-local command, so without a follow-up the + // cloud run reports completion before the model resumes the task. + this.recordTurnUsage(result.usage); + result = await this.runOwnedTurn(continueAfterCompaction); + if (messageId) { + this.pendingCompactContinuationMessageIds.delete(messageId); + } + } } + } catch (error) { + await commandSession.logWriter.flushAll(); + const { recoverable } = await this.handleTurnFailure( + commandSession.payload, + "followup", + error, + ); + if (!recoverable) { + throw error; + } + commitDelivery(); + const outcome = { stopReason: "error_recoverable" }; + resolveDelivery(outcome); + return outcome; + } finally { + this.suppressAdapterTurnComplete = false; } - } catch (error) { - if (messageId && !compactCommandCompleted) { - this.deliveredMessageIds.delete(messageId); - } - await this.session.logWriter.flushAll(); - const { recoverable } = await this.handleTurnFailure( - this.session.payload, - "followup", - error, - ); - if (!recoverable) { - throw error; - } - return { stopReason: "error_recoverable" }; - } finally { - this.suppressAdapterTurnComplete = false; - } + commitDelivery(); - this.logger.debug("User message completed", { - stopReason: result.stopReason, - }); + this.logger.debug("User message completed", { + stopReason: result.stopReason, + }); - if (result.stopReason === "end_turn") { - void this.syncCloudBranchMetadata(this.session.payload); - } + if (result.stopReason === "end_turn") { + void this.syncCloudBranchMetadata(commandSession.payload); + } - this.recordTurnUsage(result.usage); - this.broadcastTurnComplete(result.stopReason); + this.recordTurnUsage(result.usage); + this.broadcastTurnComplete(result.stopReason); - if (result.stopReason === "end_turn") { - // Relay the response to Slack. For follow-ups this is the primary - // delivery path — the HTTP caller only handles reactions. Echo the - // initiating message's id so the backend can attribute the answer. - this.relayAgentResponse(this.session.payload, messageId).catch( - (err) => - this.logger.debug("Failed to relay follow-up response", err), - ); - } + if (result.stopReason === "end_turn") { + // Relay the response to Slack. For follow-ups this is the primary + // delivery path — the HTTP caller only handles reactions. Echo the + // initiating message's id so the backend can attribute the answer. + this.relayAgentResponse(commandSession.payload, messageId).catch( + (err) => + this.logger.debug("Failed to relay follow-up response", err), + ); + } - // Flush logs and include the assistant's response text so callers - // (e.g. Slack follow-up forwarding) can extract it without racing - // against async log persistence to object storage. - let assistantMessage: string | undefined; - try { - await this.session.logWriter.flush(this.session.payload.run_id, { - coalesce: true, - }); - assistantMessage = this.session.logWriter.getFullAgentResponse( - this.session.payload.run_id, - ); - } catch { - this.logger.debug("Failed to extract assistant message from logs"); - } + // Flush logs and include the assistant's response text so callers + // (e.g. Slack follow-up forwarding) can extract it without racing + // against async log persistence to object storage. + let assistantMessage: string | undefined; + try { + await commandSession.logWriter.flush( + commandSession.payload.run_id, + { + coalesce: true, + }, + ); + assistantMessage = commandSession.logWriter.getFullAgentResponse( + commandSession.payload.run_id, + ); + } catch { + this.logger.debug("Failed to extract assistant message from logs"); + } - return { - stopReason: result.stopReason, - ...(assistantMessage && { assistant_message: assistantMessage }), - }; + const outcome = { + stopReason: result.stopReason, + ...(assistantMessage && { assistant_message: assistantMessage }), + }; + resolveDelivery(outcome); + return outcome; + } catch (error) { + if (messageId && !deliveryCommitted) { + this.deliveredMessageIds.delete(messageId); + } + rejectDelivery(error); + throw error; + } finally { + if ( + messageId && + this.inFlightMessageDeliveries.get(messageId) === deliveryOutcome + ) { + this.inFlightMessageDeliveries.delete(messageId); + } + } } case POSTHOG_NOTIFICATIONS.CANCEL: @@ -1467,10 +1556,11 @@ export class AgentServer { clientStream, ); - await clientConnection.initialize({ + const initializeResult = await clientConnection.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {}, }); + const steering = extractSteeringCapability(initializeResult); const runState = preTaskRun?.state as Record | undefined; // Preserve native Codex modes for cloud runs so they behave the same as @@ -1624,6 +1714,7 @@ export class AgentServer { runId: payload.run_id, taskId: payload.task_id, agentVersion: this.config.version ?? packageJson.version, + ...(steering ? { steering } : {}), }, }; this.broadcastEvent({ @@ -1686,6 +1777,15 @@ export class AgentServer { return { classification: classifyAgentError(message), message }; } + private async runOwnedTurn(operation: () => Promise): Promise { + this.activeOwnedTurnCount += 1; + try { + return await operation(); + } finally { + this.activeOwnedTurnCount -= 1; + } + } + /** * Send an initial/resume turn prompt, absorbing transient upstream * failures with a bounded number of retries. These turns run unattended @@ -1900,12 +2000,18 @@ export class AgentServer { }); this.session.logWriter.resetTurnMessages(payload.run_id); + const acpSessionId = this.session.acpSessionId; + if (!acpSessionId) { + throw new Error("Agent session is missing its ACP session ID"); + } - const result = await this.promptWithUpstreamRetry({ - sessionId: this.session.acpSessionId, - prompt: initialPrompt, - ...(initialPromptMeta ? { _meta: initialPromptMeta } : {}), - }); + const result = await this.runOwnedTurn(() => + this.promptWithUpstreamRetry({ + sessionId: acpSessionId, + prompt: initialPrompt, + ...(initialPromptMeta ? { _meta: initialPromptMeta } : {}), + }), + ); this.logger.debug("Initial task message completed", { stopReason: result.stopReason, @@ -2112,12 +2218,18 @@ export class AgentServer { const builtPrompt = await buildPrompt(); this.session.logWriter.resetTurnMessages(payload.run_id); + const acpSessionId = this.session.acpSessionId; + if (!acpSessionId) { + throw new Error("Agent session is missing its ACP session ID"); + } - const result = await this.promptWithUpstreamRetry({ - sessionId: this.session.acpSessionId, - prompt: builtPrompt.prompt, - ...(builtPrompt.meta ? { _meta: builtPrompt.meta } : {}), - }); + const result = await this.runOwnedTurn(() => + this.promptWithUpstreamRetry({ + sessionId: acpSessionId, + prompt: builtPrompt.prompt, + ...(builtPrompt.meta ? { _meta: builtPrompt.meta } : {}), + }), + ); this.logger.debug(`${logLabel} completed`, { stopReason: result.stopReason, diff --git a/packages/agent/src/server/schemas.ts b/packages/agent/src/server/schemas.ts index fd3f2789fe..faf79588fd 100644 --- a/packages/agent/src/server/schemas.ts +++ b/packages/agent/src/server/schemas.ts @@ -65,6 +65,8 @@ export const userMessageParamsSchema = z ]) .optional(), artifacts: z.array(z.record(z.string(), z.unknown())).optional(), + messageId: z.string().min(1).optional(), + steer: z.boolean().optional(), }) .refine( (params) => { diff --git a/packages/api-client/src/posthog-client.test.ts b/packages/api-client/src/posthog-client.test.ts index 3e6509ac08..472a8eeb74 100644 --- a/packages/api-client/src/posthog-client.test.ts +++ b/packages/api-client/src/posthog-client.test.ts @@ -1443,7 +1443,7 @@ describe("PostHogAPIClient", () => { }); }); - it("returns the entries collected so far when a later page fails", async () => { + it("marks entries collected before a failed page as incomplete", async () => { const fetch = vi .fn() .mockResolvedValueOnce(page(makeEntries(50, "a"), true)) @@ -1455,11 +1455,14 @@ describe("PostHogAPIClient", () => { }); const client = makeClient(fetch); - const result = await client.getTaskRunSessionLogs("task-1", "run-1", { - limit: 100000, - }); + const result = await client.getTaskRunSessionLogsResult( + "task-1", + "run-1", + { limit: 100000 }, + ); - expect(result).toHaveLength(50); + expect(result).toEqual({ entries: expect.any(Array), complete: false }); + expect(result.entries).toHaveLength(50); expect(fetch).toHaveBeenCalledTimes(2); }); diff --git a/packages/api-client/src/posthog-client.ts b/packages/api-client/src/posthog-client.ts index 5478769308..29c687a4fb 100644 --- a/packages/api-client/src/posthog-client.ts +++ b/packages/api-client/src/posthog-client.ts @@ -131,6 +131,11 @@ export const CLOUD_USAGE_LIMIT_ERROR_MESSAGE = "Cloud usage limit reached"; export const SESSION_LOGS_MAX_PAGE_SIZE = 5000; +export interface TaskRunSessionLogsResult { + entries: StoredLogEntry[]; + complete: boolean; +} + /** Thrown when the backend rejects a cloud run with a 429 usage-limit error. */ export class CloudUsageLimitError extends Error { limitType: UsageLimitType; @@ -3157,6 +3162,15 @@ export class PostHogAPIClient { runId: string, options?: { limit?: number; after?: string }, ): Promise { + return (await this.getTaskRunSessionLogsResult(taskId, runId, options)) + .entries; + } + + async getTaskRunSessionLogsResult( + taskId: string, + runId: string, + options?: { limit?: number; after?: string }, + ): Promise { const maxEntries = options?.limit ?? SESSION_LOGS_MAX_PAGE_SIZE; const entries: StoredLogEntry[] = []; try { @@ -3187,21 +3201,21 @@ export class PostHogAPIClient { log.warn( `Failed to fetch session logs page at offset ${offset}: ${response.status} ${response.statusText}`, ); - break; + return { entries, complete: false }; } const page = (await response.json()) as StoredLogEntry[]; entries.push(...page); const hasMore = response.headers.get("X-Has-More") === "true"; if (!hasMore || page.length === 0) { - break; + return { entries, complete: true }; } offset += page.length; } - return entries; + return { entries, complete: false }; } catch (err) { log.warn("Failed to fetch task run session logs", err); - return entries; + return { entries, complete: false }; } } diff --git a/packages/shared/src/sessions.test.ts b/packages/shared/src/sessions.test.ts index 85d736fbc3..8706578afe 100644 --- a/packages/shared/src/sessions.test.ts +++ b/packages/shared/src/sessions.test.ts @@ -95,15 +95,20 @@ describe("sessionSupportsNativeSteer", () => { { isCloud: false, steering: "interrupt-resend", adapter: "claude" }, false, ], - // Cloud runs queue/resend; they never steer locally regardless of capability. + // Cloud runs steer only when the sandbox explicitly advertises support. [ "cloud claude native", { isCloud: true, steering: "native", adapter: "claude" }, - false, + true, ], [ "cloud codex native", { isCloud: true, steering: "native", adapter: "codex" }, + true, + ], + [ + "cloud without capability", + { isCloud: true, steering: undefined, adapter: "claude" }, false, ], ])("%s", (_label, session, expected) => { diff --git a/packages/shared/src/sessions.ts b/packages/shared/src/sessions.ts index 493bb5b5ac..fd4fca99c3 100644 --- a/packages/shared/src/sessions.ts +++ b/packages/shared/src/sessions.ts @@ -62,6 +62,9 @@ export interface AgentSession { promptStartedAt: number | null; currentPromptId?: number | null; logUrl?: string; + /** Full cloud transcript entry count across the resume chain. */ + cloudTranscriptEntryCount?: number; + /** Leaf-run cursor used to reconcile live cloud log updates. */ processedLineCount?: number; framework?: "claude"; adapter?: Adapter; @@ -222,7 +225,7 @@ export function resolveBypassRevertMode( * Whether a mid-turn message can be folded into the running turn (steered) * rather than interrupt-and-resent. Decided by the adapter's negotiated * `steering` capability: "native" folds (claude, codex app-server); - * "interrupt-resend" (legacy) does not. Cloud runs never steer locally. + * "interrupt-resend" (legacy) does not. * * Fallback: if `steering` is unset (a start path that predates capability * plumbing), Claude is still treated as native — it has always steered — so the @@ -231,7 +234,7 @@ export function resolveBypassRevertMode( export function sessionSupportsNativeSteer( session: Pick, ): boolean { - if (session.isCloud) return false; if (session.steering === "native") return true; + if (session.isCloud) return false; return session.steering == null && session.adapter === "claude"; }