Merge commit 'c679822164faceeae147e26c33802c1aa3d3d0dd' into fix/codex-rejected-automation-0d3d971d

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">;