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 bf453ad651..f44fd4223f 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 @@ -32,6 +32,216 @@ import { toAgentUsage, } from "./codex-app-server-agent.js"; +describe("Codex prompt hook rejection", () => { + test("keeps the rejection visible in the manager timeline and permits a corrected submission", async () => { + const appServer = createFakeCodexAppServer(); + const provider = createProviderWithFakeAppServer(appServer); + const client: AgentClient = { + provider: "codex", + capabilities: provider.capabilities, + isAvailable: async () => true, + fetchCatalog: async () => ({ models: [], modes: [] }), + createSession: (...args) => provider.createSession(...args), + resumeSession: (...args) => provider.resumeSession(...args), + }; + const manager = new AgentManager({ clients: { codex: client }, logger: createTestLogger() }); + const agent = await manager.resumeAgentFromPersistence( + { provider: "codex", sessionId: "thread-1" }, + createConfig({ cwd: process.cwd() }), + ); + try { + 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.completeTurn(); + await failure; + + expect(manager.getTimeline(agent.id)).toEqual([ + { + type: "user_message", + text: "Disallowed prompt", + messageId: "blocked-prompt", + clientMessageId: "blocked-prompt", + }, + { type: "assistant_message", text: "[System Error] 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."); + + 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)); + appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" }); + appServer.says({ threadId: "thread-1", text: "Accepted" }); + appServer.completeTurn(); + await expect(retry).resolves.toMatchObject({ finalText: "Accepted", canceled: false }); + expect(manager.hasInFlightRun(agent.id)).toBe(false); + appServer.assertNoErrors(); + } finally { + await manager.closeAgent(agent.id); + } + }); + + test("reports the blocking reason instead of successful completion 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.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(events).toEqual([ + { + type: "turn_failed", + provider: "codex", + error: "Remove the disallowed text and retry.", + turnId: paseoTurnId, + }, + ]); + + events.length = 0; + const replacement = await session.startTurn("Allowed replacement prompt"); + appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" }); + appServer.says({ threadId: "thread-1", text: "Accepted" }); + appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(events.filter((event) => event.type.startsWith("turn_"))).toEqual([ + { type: "turn_started", provider: "codex", turnId: replacement.turnId }, + { type: "turn_completed", provider: "codex", usage: undefined, turnId: replacement.turnId }, + ]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test.each([ + { eventName: "preToolUse", status: "blocked", turnId: "native-A", threadId: "thread-1" }, + { + eventName: "userPromptSubmit", + status: "completed", + turnId: "native-A", + threadId: "thread-1", + }, + { eventName: "userPromptSubmit", status: "failed", turnId: "native-A", threadId: "thread-1" }, + { + eventName: "userPromptSubmit", + status: "blocked", + turnId: "older-turn", + threadId: "thread-1", + }, + { eventName: "userPromptSubmit", status: "blocked", turnId: null, threadId: "thread-1" }, + { + eventName: "userPromptSubmit", + status: "blocked", + turnId: "native-A", + threadId: "child-thread", + }, + ])( + "does not fail the foreground turn for an unrelated hook: $eventName $status $turnId $threadId", + async ({ eventName, status, turnId, threadId }) => { + 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, + turnId, + run: { + eventName, + status, + entries: [{ kind: "feedback", text: "Unrelated feedback" }], + }, + }, + }) + "\n", + ); + appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + expect(events).toEqual([ + { type: "turn_completed", provider: "codex", turnId: paseoTurnId, usage: undefined }, + ]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }, + ); + + test("provides an actionable failure when the blocking hook has no feedback", 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: [] }, + }, + }) + "\n", + ); + appServer.completeTurn(); + await new Promise((resolve) => setImmediate(resolve)); + expect(events).toEqual([ + { + type: "turn_failed", + provider: "codex", + turnId: paseoTurnId, + error: "Prompt rejected by a UserPromptSubmit hook.", + }, + ]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); +}); + describe("mapCodexPlanUpdateToTodo", () => { test("preserves checklist progress without creating a plan card", () => { expect( 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 bf797949c7..7bcf5069af 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 @@ -2574,6 +2574,14 @@ const CodexEventThreadRolledBackNotificationSchema = z type ParsedCodexNotification = | { kind: "thread_started"; threadId: string } | { kind: "turn_started"; turnId: string; threadId: string | null } + | { + kind: "hook_completed"; + threadId: string; + turnId: string | null; + eventName: string; + status: string; + entries: Array<{ kind: string; text: string }>; + } | { kind: "turn_completed"; status: string; @@ -2694,6 +2702,36 @@ function isCodexDeltaNotification( } const CodexNotificationSchema = z.union([ + z + .object({ + method: z.literal("hook/completed"), + params: z.object({ + threadId: z.string(), + turnId: z.string().nullish(), + run: z.object({ + eventName: z.string(), + status: z.string(), + entries: z.array(z.object({ kind: z.string(), text: z.string() })), + }), + }), + }) + .transform( + ({ params }): ParsedCodexNotification => ({ + kind: "hook_completed", + threadId: params.threadId, + turnId: params.turnId ?? null, + eventName: params.run.eventName, + status: params.run.status, + entries: params.run.entries, + }), + ), + z.object({ method: z.literal("hook/completed"), params: z.unknown() }).transform( + ({ method, params }): ParsedCodexNotification => ({ + kind: "invalid_payload", + method, + params, + }), + ), z .object({ method: z.literal("thread/started"), params: ThreadStartedNotificationSchema }) .transform( @@ -3483,6 +3521,7 @@ 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; @@ -5778,6 +5817,9 @@ export class CodexAppServerAgentSession implements AgentSession { case "turn_completed": this.handleTurnCompletedNotification(parsed); return; + case "hook_completed": + this.handleHookCompletedNotification(parsed); + return; case "plan_updated": this.handlePlanUpdatedNotification(parsed); return; @@ -6393,6 +6435,26 @@ export class CodexAppServerAgentSession implements AgentSession { this.emitEvent({ type: "thread_started", provider: CODEX_PROVIDER, sessionId: current }); } + private handleHookCompletedNotification( + parsed: Extract, + ): void { + const rejectsCurrentPrompt = + parsed.eventName === "userPromptSubmit" && + parsed.status === "blocked" && + this.currentTurnId !== null && + parsed.turnId === this.currentTurnId; + if (!rejectsCurrentPrompt) return; + + const feedback = parsed.entries + .filter( + (entry) => entry.kind === "feedback" || entry.kind === "stop" || entry.kind === "error", + ) + .map((entry) => entry.text.trim()) + .filter((text) => text.length > 0) + .join("\n\n"); + this.promptRejection = feedback || "Prompt rejected by a UserPromptSubmit hook."; + } + private handleTurnCompletedNotification( parsed: Extract, ): void { @@ -6414,11 +6476,12 @@ export class CodexAppServerAgentSession implements AgentSession { if (parsed.status !== "interrupted" && this.currentTurnId) { this.terminalProcessIdsByTurnId.delete(this.currentTurnId); } - if (parsed.status === "failed") { + // Codex completes rejected submissions successfully without accepting a user message. + if (parsed.status === "failed" || this.promptRejection !== null) { this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, - error: parsed.errorMessage ?? "Codex turn failed", + error: this.promptRejection ?? parsed.errorMessage ?? "Codex turn failed", }); } else if (parsed.status === "interrupted") { this.dismissInterruptedAsyncQuestions(); @@ -6443,6 +6506,7 @@ export class CodexAppServerAgentSession implements AgentSession { } private resetTurnTrackingState(): void { + this.promptRejection = null; this.latestPlanResult = null; this.emittedItemStartedIds.clear(); this.emittedItemCompletedIds.clear();