diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index f44fd4223f..d1deea78dd 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -33,6 +33,55 @@ import { } from "./codex-app-server-agent.js"; describe("Codex prompt hook rejection", () => { + test("shows a rejected queued follow-up without failing accepted work in the same turn", async () => { + const appServer = createFakeCodexAppServer({ + "turn/steer": () => ({ turn: { id: "native-A" } }), + }); + const { session, paseoTurnId } = await startPublicSteeringSession(appServer); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + try { + await session.steerActiveTurn!("Allowed follow-up", { expectedTurnId: paseoTurnId }); + await session.steerActiveTurn!("Rejected follow-up", { expectedTurnId: paseoTurnId }); + appServer.completesHook({ + threadId: "thread-1", + turnId: "native-A", + eventName: "userPromptSubmit", + status: "completed", + entries: [], + }); + appServer.completesHook({ + threadId: "thread-1", + turnId: "native-A", + eventName: "userPromptSubmit", + status: "blocked", + entries: [{ kind: "feedback", text: "Follow-up rejected" }], + }); + appServer.says({ threadId: "thread-1", text: "Accepted work completed" }); + const completion = waitForNextTurnCompletion(session); + appServer.completeTurn(); + await expect(completion).resolves.toMatchObject({ type: "turn_completed" }); + expect(events).toContainEqual( + expect.objectContaining({ + type: "timeline", + item: expect.objectContaining({ + type: "assistant_message", + text: "Accepted work completed", + }), + }), + ); + expect(events).toContainEqual({ + type: "timeline", + provider: "codex", + item: { type: "notification", level: "error", message: "Follow-up rejected" }, + turnId: paseoTurnId, + }); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + test("keeps the rejection visible in the manager timeline and permits a corrected submission", async () => { const appServer = createFakeCodexAppServer(); const provider = createProviderWithFakeAppServer(appServer); @@ -53,25 +102,17 @@ describe("Codex prompt hook rejection", () => { const run = manager.runAgent(agent.id, "Disallowed prompt", { clientMessageId: "blocked-prompt", }); - const failure = expect(run).rejects.toThrow("Remove the disallowed text and retry."); await appServer.waitForTurnStart(); appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" }); - appServer.child.stdout.write( - JSON.stringify({ - method: "hook/completed", - params: { - threadId: "thread-1", - turnId: "native-A", - run: { - eventName: "userPromptSubmit", - status: "blocked", - entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }], - }, - }, - }) + "\n", - ); + appServer.completesHook({ + threadId: "thread-1", + turnId: "native-A", + eventName: "userPromptSubmit", + status: "blocked", + entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }], + }); appServer.completeTurn(); - await failure; + await expect(run).resolves.toMatchObject({ finalText: "", canceled: false }); expect(manager.getTimeline(agent.id)).toEqual([ { @@ -80,18 +121,15 @@ describe("Codex prompt hook rejection", () => { messageId: "blocked-prompt", clientMessageId: "blocked-prompt", }, - { type: "assistant_message", text: "[System Error] Remove the disallowed text and retry." }, + { type: "notification", level: "error", message: "Remove the disallowed text and retry." }, ]); expect(manager.hasInFlightRun(agent.id)).toBe(false); - expect(manager.getAgent(agent.id)?.lastError).toBe("Remove the disallowed text and retry."); + expect(manager.getAgent(agent.id)?.lastError).toBeUndefined(); const retry = manager.runAgent(agent.id, "Corrected prompt", { clientMessageId: "corrected-prompt", }); - function isTurnStart(request: Record): boolean { - return request.method === "turn/start"; - } - await vi.waitFor(() => expect(appServer.requests().filter(isTurnStart)).toHaveLength(2)); + await appServer.waitForTurnStart(2); appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" }); appServer.says({ threadId: "thread-1", text: "Accepted" }); appServer.completeTurn(); @@ -103,35 +141,39 @@ describe("Codex prompt hook rejection", () => { } }); - test("reports the blocking reason instead of successful completion and accepts the next turn", async () => { + test("shows blocking feedback and accepts the next turn", async () => { const appServer = createFakeCodexAppServer(); const { session, paseoTurnId } = await startPublicSteeringSession(appServer); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); try { - appServer.child.stdout.write( - JSON.stringify({ - method: "hook/completed", - params: { - threadId: "thread-1", - turnId: "native-A", - run: { - eventName: "userPromptSubmit", - status: "blocked", - entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }], - }, - }, - }) + "\n", - ); + appServer.completesHook({ + threadId: "thread-1", + turnId: "native-A", + eventName: "userPromptSubmit", + status: "blocked", + entries: [{ kind: "feedback", text: "Remove the disallowed text and retry." }], + }); + const completion = waitForNextTurnCompletion(session); appServer.completeTurn(); - await new Promise((resolve) => setImmediate(resolve)); + await completion; expect(events).toEqual([ { - type: "turn_failed", + type: "timeline", provider: "codex", - error: "Remove the disallowed text and retry.", + item: { + type: "notification", + level: "error", + message: "Remove the disallowed text and retry.", + }, + turnId: paseoTurnId, + }, + { + type: "turn_completed", + provider: "codex", + usage: undefined, turnId: paseoTurnId, }, ]); @@ -140,8 +182,9 @@ describe("Codex prompt hook rejection", () => { const replacement = await session.startTurn("Allowed replacement prompt"); appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" }); appServer.says({ threadId: "thread-1", text: "Accepted" }); + const replacementCompletion = waitForNextTurnCompletion(session); appServer.completeTurn(); - await new Promise((resolve) => setImmediate(resolve)); + await replacementCompletion; expect(events.filter((event) => event.type.startsWith("turn_"))).toEqual([ { type: "turn_started", provider: "codex", turnId: replacement.turnId }, @@ -183,22 +226,16 @@ describe("Codex prompt hook rejection", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); try { - appServer.child.stdout.write( - JSON.stringify({ - method: "hook/completed", - params: { - threadId, - turnId, - run: { - eventName, - status, - entries: [{ kind: "feedback", text: "Unrelated feedback" }], - }, - }, - }) + "\n", - ); + appServer.completesHook({ + threadId, + turnId, + eventName, + status, + entries: [{ kind: "feedback", text: "Unrelated feedback" }], + }); + const completion = waitForNextTurnCompletion(session); appServer.completeTurn(); - await new Promise((resolve) => setImmediate(resolve)); + await completion; expect(events).toEqual([ { type: "turn_completed", provider: "codex", turnId: paseoTurnId, usage: undefined }, ]); @@ -215,24 +252,32 @@ describe("Codex prompt hook rejection", () => { const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); try { - appServer.child.stdout.write( - JSON.stringify({ - method: "hook/completed", - params: { - threadId: "thread-1", - turnId: "native-A", - run: { eventName: "userPromptSubmit", status: "blocked", entries: [] }, - }, - }) + "\n", - ); + appServer.completesHook({ + threadId: "thread-1", + turnId: "native-A", + eventName: "userPromptSubmit", + status: "blocked", + entries: [], + }); + const completion = waitForNextTurnCompletion(session); appServer.completeTurn(); - await new Promise((resolve) => setImmediate(resolve)); + await completion; expect(events).toEqual([ { - type: "turn_failed", + type: "timeline", provider: "codex", + item: { + type: "notification", + level: "error", + message: "Prompt rejected by a UserPromptSubmit hook.", + }, + turnId: paseoTurnId, + }, + { + type: "turn_completed", + provider: "codex", + usage: undefined, turnId: paseoTurnId, - error: "Prompt rejected by a UserPromptSubmit hook.", }, ]); appServer.assertNoErrors(); @@ -301,6 +346,7 @@ import { fakeCodexThreadHistory, type FakeCodexAppServer, waitForNextEvent, + waitForNextTurnCompletion, waitForNextPermission, waitForNextTimelineItem, waitForProviderSubagent, diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 7bcf5069af..83691d9e2a 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -3521,7 +3521,6 @@ export class CodexAppServerAgentSession implements AgentSession { // in this app-server, so no other app-server can resume it and it is not reported. private savedThreadId: string | null = null; private currentTurnId: string | null = null; - private promptRejection: string | null = null; private pendingForegroundTurnIdentification: { foregroundTurnId: string; promise: Promise; @@ -6452,7 +6451,15 @@ export class CodexAppServerAgentSession implements AgentSession { .map((entry) => entry.text.trim()) .filter((text) => text.length > 0) .join("\n\n"); - this.promptRejection = feedback || "Prompt rejected by a UserPromptSubmit hook."; + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: { + type: "notification", + level: "error", + message: feedback || "Prompt rejected by a UserPromptSubmit hook.", + }, + }); } private handleTurnCompletedNotification( @@ -6476,12 +6483,11 @@ export class CodexAppServerAgentSession implements AgentSession { if (parsed.status !== "interrupted" && this.currentTurnId) { this.terminalProcessIdsByTurnId.delete(this.currentTurnId); } - // Codex completes rejected submissions successfully without accepting a user message. - if (parsed.status === "failed" || this.promptRejection !== null) { + if (parsed.status === "failed") { this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, - error: this.promptRejection ?? parsed.errorMessage ?? "Codex turn failed", + error: parsed.errorMessage ?? "Codex turn failed", }); } else if (parsed.status === "interrupted") { this.dismissInterruptedAsyncQuestions(); @@ -6506,7 +6512,6 @@ export class CodexAppServerAgentSession implements AgentSession { } private resetTurnTrackingState(): void { - this.promptRejection = null; this.latestPlanResult = null; this.emittedItemStartedIds.clear(); this.emittedItemCompletedIds.clear(); diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index ec14a773e4..0b931bb1c7 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -56,6 +56,13 @@ interface FakeLegacyPatch { path: string; diff: string; } +interface FakeHookCompletion { + threadId: string; + turnId: string | null; + eventName: string; + status: string; + entries: Array<{ kind: string; text: string }>; +} type CodexAppServerChildProcess = ChildProcessWithoutNullStreams & { stdin: PassThrough; stdout: PassThrough; @@ -67,13 +74,14 @@ export interface FakeCodexAppServer { readonly recordedRollbacks: JsonObject[]; requests(): readonly JsonObject[]; assertNoErrors(): void; - waitForTurnStart(): Promise; + waitForTurnStart(occurrence?: number): Promise; waitForRequest(method: string): Promise; disconnect(): void; nextResponse(): Promise; startsTurn(params: { threadId: string; turnId?: string }): void; startsCompaction(params: { threadId: string; itemId: string }): void; updatesPlan(params: { threadId: string; steps: string[] }): void; + completesHook(params: FakeHookCompletion): void; completeTurn(params?: { threadId?: string; status?: "completed" | "failed" | "interrupted"; @@ -368,9 +376,11 @@ export function createFakeCodexAppServer( throw errors[0]; } }, - async waitForTurnStart() { + async waitForTurnStart(occurrence = 1) { const message = await waitForMessage( - (candidate) => candidate.method === "turn/start", + (candidate) => + messages.filter((request) => request.method === "turn/start")[occurrence - 1] === + candidate, "turn start request", ); return toJsonObject(message.params); @@ -418,6 +428,9 @@ export function createFakeCodexAppServer( })}\n`, ); }, + completesHook({ threadId, turnId, ...run }) { + writeNotification("hook/completed", { threadId, turnId, run }); + }, completeTurn(params = {}) { child.stdout.write( `${JSON.stringify({ @@ -689,16 +702,17 @@ type StreamEventOfType = Extract( session: AgentSession, - type: TType, + type: TType | readonly TType[], accepts?: (event: StreamEventOfType) => boolean, ): Promise> { + const types: readonly StreamEventType[] = typeof type === "string" ? [type] : type; return new Promise((resolve, reject) => { const timeout = setTimeout(() => { unsubscribe(); - reject(new Error(`Timed out waiting for ${type}`)); + reject(new Error(`Timed out waiting for ${types.join(" or ")}`)); }, 1000); const unsubscribe = session.subscribe((event) => { - if (event.type !== type) { + if (!types.includes(event.type)) { return; } const typedEvent = event as StreamEventOfType; @@ -712,6 +726,10 @@ export function waitForNextEvent( }); } +export function waitForNextTurnCompletion(session: AgentSession) { + return waitForNextEvent(session, ["turn_completed", "turn_failed", "turn_canceled"]); +} + type TimelineEvent = StreamEventOfType<"timeline">; type ProviderSubagentEvent = StreamEventOfType<"provider_subagent">;