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..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 @@ -32,6 +32,261 @@ import { toAgentUsage, } 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); + 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", + }); + await appServer.waitForTurnStart(); + appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" }); + 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 expect(run).resolves.toMatchObject({ finalText: "", canceled: false }); + + expect(manager.getTimeline(agent.id)).toEqual([ + { + type: "user_message", + text: "Disallowed prompt", + messageId: "blocked-prompt", + clientMessageId: "blocked-prompt", + }, + { type: "notification", level: "error", message: "Remove the disallowed text and retry." }, + ]); + expect(manager.hasInFlightRun(agent.id)).toBe(false); + expect(manager.getAgent(agent.id)?.lastError).toBeUndefined(); + + const retry = manager.runAgent(agent.id, "Corrected prompt", { + clientMessageId: "corrected-prompt", + }); + await appServer.waitForTurnStart(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("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.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 completion; + + expect(events).toEqual([ + { + type: "timeline", + provider: "codex", + item: { + type: "notification", + level: "error", + message: "Remove the disallowed text and retry.", + }, + turnId: paseoTurnId, + }, + { + type: "turn_completed", + provider: "codex", + usage: undefined, + 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" }); + const replacementCompletion = waitForNextTurnCompletion(session); + appServer.completeTurn(); + await replacementCompletion; + + 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.completesHook({ + threadId, + turnId, + eventName, + status, + entries: [{ kind: "feedback", text: "Unrelated feedback" }], + }); + const completion = waitForNextTurnCompletion(session); + appServer.completeTurn(); + await completion; + 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.completesHook({ + threadId: "thread-1", + turnId: "native-A", + eventName: "userPromptSubmit", + status: "blocked", + entries: [], + }); + const completion = waitForNextTurnCompletion(session); + appServer.completeTurn(); + await completion; + expect(events).toEqual([ + { + 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, + }, + ]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); +}); + describe("mapCodexPlanUpdateToTodo", () => { test("preserves checklist progress without creating a plan card", () => { expect( @@ -91,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 bf797949c7..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 @@ -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( @@ -5778,6 +5816,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 +6434,34 @@ 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.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: { + type: "notification", + level: "error", + message: feedback || "Prompt rejected by a UserPromptSubmit hook.", + }, + }); + } + private handleTurnCompletedNotification( parsed: Extract, ): void { 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">;