fix(codex): show hook rejections without failing accepted work

A blocked queued submission can share a native turn with accepted input.
Show its feedback through the existing error notification instead of
replacing Codex's successful turn outcome. Keep protocol framing and
completion waits inside the existing test fixture.

Co-authored-by: Scott Zhang <scottzh@fireworks.ai>

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,

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<ParsedCodexNotification, { kind: "hook_completed" }>,
+  ): 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<ParsedCodexNotification, { kind: "turn_completed" }>,
   ): void {

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<JsonObject>;
+  waitForTurnStart(occurrence?: number): Promise<JsonObject>;
   waitForRequest(method: string): Promise<JsonObject>;
   disconnect(): void;
   nextResponse(): Promise<string>;
   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<TType extends StreamEventType> = Extract<AgentStreamEvent
 
 export function waitForNextEvent<TType extends StreamEventType>(
   session: AgentSession,
-  type: TType,
+  type: TType | readonly TType[],
   accepts?: (event: StreamEventOfType<TType>) => boolean,
 ): Promise<StreamEventOfType<TType>> {
+  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<TType>;
@@ -712,6 +726,10 @@ export function waitForNextEvent<TType extends StreamEventType>(
   });
 }
 
+export function waitForNextTurnCompletion(session: AgentSession) {
+  return waitForNextEvent(session, ["turn_completed", "turn_failed", "turn_canceled"]);
+}
+
 type TimelineEvent = StreamEventOfType<"timeline">;
 type ProviderSubagentEvent = StreamEventOfType<"provider_subagent">;