Show Codex reconnects and warnings without false recovery on child completion (#9370)

Show Codex reconnect attempts and distinct warnings through existing timeline rows. Keep interrupted root retries running through child activity and settle them on recovery, turn end, session close, or process exit. Closes #9236 Co-authored-by: whatamelon <39753493+whatamelon@users.noreply.github.com>
2026-11-11 10:57:07 +00:00 · 2026-11-11 10:10:32 +00:00
parent a29e06bf1b
commit aade09d056
4 changed files with 559 additions and 9 deletions
@@ -171,7 +171,11 @@ async function startMockResponsesServer(sequence: string[]): Promise<{
  };
 }

-function writeMockCodexConfig(codexHome: string, serverUrl: string): void {
+function writeMockCodexConfig(
+  codexHome: string,
+  serverUrl: string,
+  options: { streamMaxRetries?: number } = {},
+): void {
  writeFileSync(
    path.join(codexHome, "config.toml"),
    `
@@ -186,7 +190,7 @@ name = "Mock provider for test"
 base_url = "${serverUrl}/v1"
 wire_api = "responses"
 request_max_retries = 0
-stream_max_retries = 0
+stream_max_retries = ${options.streamMaxRetries ?? 0}
 `,
  );
 }
@@ -546,4 +550,78 @@ describe("Codex app-server provider (local e2e)", () => {
    },
    30_000,
  );
+
+  test.runIf(isCodexInstalled())(
+    "shows Codex reconnecting a dropped model stream until the reply arrives",
+    async () => {
+      const cwd = mkdtempSync(path.join(os.tmpdir(), "codex-reconnect-cwd-"));
+      const codexHome = mkdtempSync(path.join(os.tmpdir(), "codex-reconnect-home-"));
+      // A stream that ends before `response.completed` makes Codex reconnect.
+      const droppedStream = sse([responseCreated("resp-dropped")]);
+      const mockServer = await startMockResponsesServer([
+        droppedStream,
+        droppedStream,
+        assistantMessageSse("recovered"),
+      ]);
+
+      try {
+        writeMockCodexConfig(codexHome, mockServer.url, { streamMaxRetries: 2 });
+        const client = new CodexAppServerAgentClient(createTestLogger());
+        const session = await client.createSession(
+          { provider: "codex", cwd, modeId: "auto", model: "mock-model" },
+          { env: { CODEX_HOME: codexHome } },
+        );
+
+        try {
+          const events: AgentStreamEvent[] = [];
+          session.subscribe((event) => {
+            events.push(event);
+          });
+          const turnFinished = waitForEvent({
+            session,
+            timeoutMs: 20_000,
+            label: "turn completion",
+            predicate: (
+              event,
+            ): event is Extract<
+              AgentStreamEvent,
+              { type: "turn_completed" | "turn_failed" | "turn_canceled" }
+            > =>
+              event.type === "turn_completed" ||
+              event.type === "turn_failed" ||
+              event.type === "turn_canceled",
+          });
+
+          await session.startTurn("reply after reconnecting");
+          expect((await turnFinished).type).toBe("turn_completed");
+
+          const retryRows = events.flatMap((event) =>
+            event.type === "timeline" &&
+            event.item.type === "tool_call" &&
+            event.item.name === "codex_retry"
+              ? [event.item]
+              : [],
+          );
+          expect(
+            retryRows.map(
+              (row) => `${row.status}: ${row.detail.type === "plain_text" ? row.detail.label : ""}`,
+            ),
+          ).toEqual([
+            "running: Reconnecting... 1/2",
+            "running: Reconnecting... 2/2",
+            "completed: Reconnected",
+          ]);
+          expect(new Set(retryRows.map((row) => row.callId)).size).toBe(1);
+          expect(mockServer.requestBodies).toHaveLength(3);
+        } finally {
+          await session.close();
+        }
+      } finally {
+        await mockServer.close();
+        rmSync(cwd, { recursive: true, force: true });
+        rmSync(codexHome, { recursive: true, force: true });
+      }
+    },
+    30_000,
+  );
 });
@@ -603,6 +603,58 @@ async function startCompactionTurnTest(): Promise<{
  return { appServer, session, events, terminalEvent };
 }

+async function startStatusNoticeTurnTest(): Promise<{
+  appServer: FakeCodexAppServer;
+  session: AgentSession;
+  events: AgentStreamEvent[];
+  terminalEvent: Promise<TurnTerminalEvent>;
+}> {
+  const appServer = createFakeCodexAppServer();
+  const session = new CodexAppServerAgentSession(
+    createConfig({ cwd: "/workspace/project" }),
+    null,
+    createTestLogger(),
+    async () => appServer.child,
+  );
+  const events: AgentStreamEvent[] = [];
+  const terminalEvent = new Promise<TurnTerminalEvent>((resolve) => {
+    session.subscribe((event) => {
+      const isStatusNotice =
+        event.type === "timeline" &&
+        (event.item.type === "notification" ||
+          (event.item.type === "tool_call" && event.item.name === "codex_retry"));
+      const isTerminal =
+        event.type === "turn_completed" ||
+        event.type === "turn_failed" ||
+        event.type === "turn_canceled";
+      if (isStatusNotice || isTerminal) {
+        events.push(event);
+      }
+      if (isTerminal) {
+        resolve(event);
+      }
+    });
+  });
+
+  await session.startTurn("exercise status notices");
+  appServer.startsTurn({ threadId: "thread-1", turnId: "codex-turn-1" });
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  return { appServer, session, events, terminalEvent };
+}
+
+function describeStatusNotice(event: AgentStreamEvent): string {
+  if (event.type !== "timeline") {
+    return event.type;
+  }
+  if (event.item.type === "notification") {
+    return `${event.item.level}: ${event.item.message}`;
+  }
+  if (event.item.type === "tool_call" && event.item.detail.type === "plain_text") {
+    return `${event.item.status}: ${event.item.detail.label}`;
+  }
+  return event.item.type;
+}
+
 function archivedThreadHandle() {
  return {
    sessionId: "archived-thread-id",
@@ -6223,6 +6275,229 @@ describe("Codex app-server provider", () => {
    ]);
  });

+  test("shows Codex stream reconnects as one status line until the reply resumes", async () => {
+    const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest();
+
+    try {
+      for (const attempt of ["2/5", "3/5"]) {
+        appServer.reportsTurnError({
+          threadId: "thread-1",
+          turnId: "codex-turn-1",
+          message: `Reconnecting... ${attempt}`,
+          additionalDetails: "stream disconnected before completion: idle timeout waiting for SSE",
+          willRetry: true,
+        });
+      }
+      // A message steered into the turn while it waits is not the stream coming back.
+      emitCodexUserMessage(appServer, {
+        id: "steer",
+        text: "Still there?",
+        turnId: "codex-turn-1",
+      });
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(events.map(describeStatusNotice)).toEqual([
+        "running: Reconnecting... 2/5",
+        "running: Reconnecting... 3/5",
+      ]);
+
+      appServer.says({ threadId: "thread-1", itemId: "reply", text: "Back online." });
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      appServer.completeTurn();
+      await terminalEvent;
+
+      expect(events.map(describeStatusNotice)).toEqual([
+        "running: Reconnecting... 2/5",
+        "running: Reconnecting... 3/5",
+        "completed: Reconnected",
+        "turn_completed",
+      ]);
+      const retryRows = events.flatMap((event) =>
+        event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [],
+      );
+      expect(new Set(retryRows.map((row) => row.callId)).size).toBe(1);
+      expect(retryRows[1]?.detail).toEqual({
+        type: "plain_text",
+        label: "Reconnecting... 3/5",
+        text: "stream disconnected before completion: idle timeout waiting for SSE",
+        icon: "sparkles",
+      });
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("keeps one root reconnect row while a child completes", async () => {
+    const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest();
+    try {
+      appServer.reportsTurnError({
+        threadId: "thread-1",
+        turnId: "codex-turn-1",
+        message: "Reconnecting... 1/2",
+        willRetry: true,
+      });
+      // Codex reports a child's completion on the root thread, independently
+      // of whether the root model stream has recovered.
+      appServer.beginsSubAgentActivity({
+        callId: "subagent-completed-child-turn",
+        threadId: "child-thread",
+        parentThreadId: "thread-1",
+        agentPath: "/root/child",
+        kind: "completed",
+      });
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(events.map(describeStatusNotice)).toEqual(["running: Reconnecting... 1/2"]);
+
+      appServer.reportsTurnError({
+        threadId: "thread-1",
+        turnId: "codex-turn-1",
+        message: "Reconnecting... 2/2",
+        willRetry: true,
+      });
+      appServer.says({ threadId: "thread-1", itemId: "reply", text: "Root recovered." });
+      appServer.completeTurn();
+      await terminalEvent;
+      expect(events.map(describeStatusNotice)).toEqual([
+        "running: Reconnecting... 1/2",
+        "running: Reconnecting... 2/2",
+        "completed: Reconnected",
+        "turn_completed",
+      ]);
+      const rows = events.flatMap((event) =>
+        event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [],
+      );
+      expect(new Set(rows.map((row) => row.callId)).size).toBe(1);
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("fails queued reconnect notices before reporting process disconnection", async () => {
+    const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest();
+    try {
+      for (const attempt of ["1/2", "2/2"]) {
+        appServer.reportsTurnError({
+          threadId: "thread-1",
+          turnId: "codex-turn-1",
+          message: `Reconnecting... ${attempt}`,
+          willRetry: true,
+        });
+      }
+      // Exit before the session's ordered notification queue has drained.
+      appServer.disconnect();
+      await terminalEvent;
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(events.map(describeStatusNotice)).toEqual([
+        "running: Reconnecting... 1/2",
+        "running: Reconnecting... 2/2",
+        "failed: Reconnecting... 2/2",
+        "turn_failed",
+      ]);
+      const rows = events.flatMap((event) =>
+        event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [],
+      );
+      expect(new Set(rows.map((row) => row.callId)).size).toBe(1);
+      expect(new Set(events.map((event) => event.turnId)).size).toBe(1);
+      expect(events[0]?.turnId).toBeTruthy();
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("cancels the reconnect row before closing its session", async () => {
+    const { appServer, session, events } = await startStatusNoticeTurnTest();
+    try {
+      appServer.reportsTurnError({
+        threadId: "thread-1",
+        turnId: "codex-turn-1",
+        message: "Reconnecting... 1/2",
+        willRetry: true,
+      });
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      await session.close();
+      expect(events.map(describeStatusNotice)).toEqual([
+        "running: Reconnecting... 1/2",
+        "canceled: Reconnecting... 1/2",
+      ]);
+      const rows = events.flatMap((event) =>
+        event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [],
+      );
+      expect(new Set(rows.map((row) => row.callId)).size).toBe(1);
+      // Closing again must not publish another terminal update.
+      await session.close();
+      expect(events).toHaveLength(2);
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("fails the reconnect status line with the turn when Codex stops retrying", async () => {
+    const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest();
+
+    try {
+      appServer.reportsTurnError({
+        threadId: "thread-1",
+        turnId: "codex-turn-1",
+        message: "Reconnecting... 5/5",
+        additionalDetails: "stream disconnected before completion: idle timeout waiting for SSE",
+        willRetry: true,
+      });
+      appServer.reportsTurnError({
+        threadId: "thread-1",
+        turnId: "codex-turn-1",
+        message: "stream disconnected before completion: idle timeout waiting for SSE",
+        willRetry: false,
+      });
+      appServer.completeTurn({
+        status: "failed",
+        error: { message: "stream disconnected before completion: idle timeout waiting for SSE" },
+      });
+      await terminalEvent;
+
+      expect(events.map(describeStatusNotice)).toEqual([
+        "running: Reconnecting... 5/5",
+        "failed: Reconnecting... 5/5",
+        "turn_failed",
+      ]);
+      expect(events[1]).toMatchObject({
+        item: { error: "stream disconnected before completion: idle timeout waiting for SSE" },
+      });
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("shows each Codex warning once as a notice in the thread", async () => {
+    const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest();
+    const missingMetadata =
+      "Model metadata for `mock-model` not found. Defaulting to fallback metadata; this can degrade performance and cause issues.";
+
+    try {
+      appServer.warns({ threadId: "thread-1", message: missingMetadata });
+      appServer.warns({
+        threadId: null,
+        message: "Falling back from WebSockets to HTTPS transport.",
+      });
+      // Codex repeats the missing-metadata warning on every turn.
+      appServer.warns({ threadId: "thread-1", message: missingMetadata });
+      appServer.completeTurn();
+      await terminalEvent;
+
+      expect(events.map(describeStatusNotice)).toEqual([
+        `warning: ${missingMetadata}`,
+        "warning: Falling back from WebSockets to HTTPS transport.",
+        "turn_completed",
+      ]);
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
  test("does not replace a persisted Codex thread when app-server resume fails", async () => {
    const session = createSession({ thinkingOptionId: "medium" });
    session.currentThreadId = "archived-thread-id";
@@ -1200,6 +1200,40 @@ export function mapCodexPlanToToolCall(params: {
  };
 }

+interface CodexStreamRetry {
+  callId: string;
+  message: string;
+  details: string | null;
+}
+
+type CodexStreamRetryOutcome =
+  | { status: "running" | "completed" | "canceled" }
+  | { status: "failed"; error: string };
+
+// Codex reports each model-stream reconnect as an `error` notification with `willRetry: true`.
+// One status line per reconnect run shows the latest attempt and settles when the turn moves on.
+function mapCodexStreamRetryToStatusLine(
+  retry: CodexStreamRetry,
+  outcome: CodexStreamRetryOutcome,
+): ToolCallTimelineItem {
+  const base = {
+    type: "tool_call" as const,
+    callId: retry.callId,
+    name: "codex_retry",
+    detail: {
+      type: "plain_text" as const,
+      label: outcome.status === "completed" ? "Reconnected" : retry.message,
+      ...(retry.details ? { text: retry.details } : {}),
+      icon: "sparkles" as const,
+    },
+    metadata: { synthetic: true, source: "codex_stream_retry" },
+  };
+  if (outcome.status === "failed") {
+    return { ...base, status: "failed", error: outcome.error };
+  }
+  return { ...base, status: outcome.status, error: null };
+}
+
 function buildPlanPermissionActions(options?: {
  includeResumeAction?: boolean;
  resumeLabel?: string;
@@ -2323,6 +2357,26 @@ const TurnCompletedNotificationSchema = z
  })
  .passthrough();

+const TurnErrorNotificationSchema = z
+  .object({
+    threadId: z.string().optional(),
+    willRetry: z.boolean(),
+    error: z
+      .object({
+        message: z.string(),
+        additionalDetails: z.string().nullable().optional(),
+      })
+      .passthrough(),
+  })
+  .passthrough();
+
+const WarningNotificationSchema = z
+  .object({
+    threadId: z.string().nullable().optional(),
+    message: z.string(),
+  })
+  .passthrough();
+
 const TurnPlanUpdatedNotificationSchema = z
  .object({
    threadId: z.string().optional(),
@@ -2605,6 +2659,14 @@ type ParsedCodexNotification =
      errorMessage: string | null;
      threadId: string | null;
    }
+  | {
+      kind: "turn_error";
+      message: string;
+      details: string | null;
+      willRetry: boolean;
+      threadId: string | null;
+    }
+  | { kind: "warning"; message: string; threadId: string | null }
  | {
      kind: "plan_updated";
      plan: Array<{ step: string | null; status: string | null }>;
@@ -2765,6 +2827,36 @@ const CodexNotificationSchema = z.union([
      params,
    }),
  ),
+  z.object({ method: z.literal("error"), params: TurnErrorNotificationSchema }).transform(
+    ({ params }): ParsedCodexNotification => ({
+      kind: "turn_error",
+      message: params.error.message,
+      details: params.error.additionalDetails ?? null,
+      willRetry: params.willRetry,
+      threadId: params.threadId ?? null,
+    }),
+  ),
+  z.object({ method: z.literal("error"), params: z.unknown() }).transform(
+    ({ method, params }): ParsedCodexNotification => ({
+      kind: "invalid_payload",
+      method,
+      params,
+    }),
+  ),
+  z.object({ method: z.literal("warning"), params: WarningNotificationSchema }).transform(
+    ({ params }): ParsedCodexNotification => ({
+      kind: "warning",
+      message: params.message,
+      threadId: params.threadId ?? null,
+    }),
+  ),
+  z.object({ method: z.literal("warning"), params: z.unknown() }).transform(
+    ({ method, params }): ParsedCodexNotification => ({
+      kind: "invalid_payload",
+      method,
+      params,
+    }),
+  ),
  z
    .object({ method: z.literal("turn/plan/updated"), params: TurnPlanUpdatedNotificationSchema })
    .transform(
@@ -3558,6 +3650,9 @@ export class CodexAppServerAgentSession implements AgentSession {
  private pendingFileChangeOutputDeltas = new Map<string, string[]>();
  private pendingToolCallDetails = new Map<string, ToolCallTimelineItem["detail"]>();
  private pendingAssistantMessageBoundary = false;
+  // The root turn's model-stream reconnect run that has not yet resumed or ended.
+  private activeStreamRetry: CodexStreamRetry | null = null;
+  private readonly shownWarningMessages = new Set<string>();
  private terminalCommandByProcessId = new Map<string, string>();
  // The root thread's still-running command process ids, keyed by the native turn that
  // started them. Codex's `thread/backgroundTerminals/list` carries no turn, so
@@ -3684,6 +3779,8 @@ export class CodexAppServerAgentSession implements AgentSession {
  }

  async connect(): Promise<void> {
+    // A replacement process must not inherit the previous process's queued events.
+    if (this.connectionState === "disconnected") await this.pendingNotifications;
    if (this.closed) {
      throw this.createClosedError();
    }
@@ -3731,7 +3828,7 @@ export class CodexAppServerAgentSession implements AgentSession {
    }
    this.client = client;
    client.setUnexpectedTerminationHandler((error) => {
-      this.handleUnexpectedTermination(error);
+      this.pendingNotifications = this.queueUnexpectedTermination(error);
    });
    client.setNotificationHandler((method, params) => this.handleNotification(method, params));
    this.registerRequestHandlers();
@@ -3807,12 +3904,22 @@ export class CodexAppServerAgentSession implements AgentSession {
    };
  }

+  private async queueUnexpectedTermination(error: Error): Promise<void> {
+    this.connectionState = "disconnected";
+    this.background.stop();
+    // Finish notifications already received from this process before ending
+    // their turn, so queued retry notices cannot reopen a failed retry row.
+    await this.pendingNotifications;
+    if (!this.closed) this.handleUnexpectedTermination(error);
+  }
+
  private handleUnexpectedTermination(error: Error): void {
    this.connectionState = "disconnected";
    this.background.stop();
    const hasActiveRootTurn = this.activeForegroundTurnId !== null || this.currentTurnId !== null;
    this.clearPendingPermissions({ preservePlanApprovals: !hasActiveRootTurn });
    if (hasActiveRootTurn) {
+      this.settleStreamRetry({ status: "failed", error: error.message });
      this.emitEvent({
        type: "turn_failed",
        provider: CODEX_PROVIDER,
@@ -5396,6 +5503,7 @@ export class CodexAppServerAgentSession implements AgentSession {
    this.clearPendingPermissions();
    this.pendingToolCallDetails.clear();
    this.pendingSubAgentNotificationsByThreadId.clear();
+    this.settleStreamRetry({ status: "canceled" });
    this.subscribers.clear();
    this.startupEvents = null;
    this.activeForegroundTurnId = null;
@@ -5823,6 +5931,19 @@ export class CodexAppServerAgentSession implements AgentSession {
      await this.dispatchSubAgentNotification(parsed, route.callId);
      return;
    }
+    // Root model output means the stream is back. Steering messages and asynchronous
+    // child activity arrive independently of the root stream, so neither proves recovery.
+    const startedItemType =
+      parsed.kind === "item_started" ? normalizeCodexThreadItemType(parsed.item.type) : null;
+    if (
+      parsed.kind === "agent_message_delta" ||
+      parsed.kind === "reasoning_delta" ||
+      (parsed.kind === "item_started" &&
+        startedItemType !== "userMessage" &&
+        startedItemType !== "subAgentActivity")
+    ) {
+      this.settleStreamRetry({ status: "completed" });
+    }
    if (
      parsed.kind === "turn_completed" ||
      parsed.kind === "exec_command_started" ||
@@ -5895,6 +6016,12 @@ export class CodexAppServerAgentSession implements AgentSession {
      case "turn_completed":
        this.handleTurnCompletedNotification(parsed);
        return;
+      case "turn_error":
+        this.handleTurnErrorNotification(parsed);
+        return;
+      case "warning":
+        this.handleWarningNotification(parsed);
+        return;
      case "plan_updated":
        this.handlePlanUpdatedNotification(parsed);
        return;
@@ -6533,15 +6660,15 @@ export class CodexAppServerAgentSession implements AgentSession {
      this.terminalProcessIdsByTurnId.delete(this.currentTurnId);
    }
    if (parsed.status === "failed") {
-      this.emitEvent({
-        type: "turn_failed",
-        provider: CODEX_PROVIDER,
-        error: parsed.errorMessage ?? "Codex turn failed",
-      });
+      const error = parsed.errorMessage ?? "Codex turn failed";
+      this.settleStreamRetry({ status: "failed", error });
+      this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, error });
    } else if (parsed.status === "interrupted") {
+      this.settleStreamRetry({ status: "canceled" });
      this.dismissInterruptedAsyncQuestions();
      this.emitEvent({ type: "turn_canceled", provider: CODEX_PROVIDER, reason: "interrupted" });
    } else {
+      this.settleStreamRetry({ status: "completed" });
      if (this.planModeEnabled && this.latestPlanResult?.text) {
        this.emitSyntheticPlanApprovalRequest(this.latestPlanResult.text);
      }
@@ -6560,6 +6687,53 @@ export class CodexAppServerAgentSession implements AgentSession {
    this.resetTurnTrackingState();
  }

+  private handleTurnErrorNotification(
+    parsed: Extract<ParsedCodexNotification, { kind: "turn_error" }>,
+  ): void {
+    // A final error is followed by `turn/completed`, which reports it as the turn's failure.
+    if (!parsed.willRetry) {
+      return;
+    }
+    this.activeStreamRetry = {
+      callId: this.activeStreamRetry?.callId ?? `codex-retry:${randomUUID()}`,
+      message: parsed.message,
+      details: parsed.details,
+    };
+    this.emitEvent({
+      type: "timeline",
+      provider: CODEX_PROVIDER,
+      item: mapCodexStreamRetryToStatusLine(this.activeStreamRetry, { status: "running" }),
+    });
+  }
+
+  private handleWarningNotification(
+    parsed: Extract<ParsedCodexNotification, { kind: "warning" }>,
+  ): void {
+    // Codex repeats some warnings on every turn, such as missing metadata for a custom model.
+    if (this.shownWarningMessages.has(parsed.message)) {
+      return;
+    }
+    this.shownWarningMessages.add(parsed.message);
+    this.emitEvent({
+      type: "timeline",
+      provider: CODEX_PROVIDER,
+      item: { type: "notification", level: "warning", message: parsed.message },
+    });
+  }
+
+  private settleStreamRetry(outcome: CodexStreamRetryOutcome): void {
+    const retry = this.activeStreamRetry;
+    if (!retry) {
+      return;
+    }
+    this.activeStreamRetry = null;
+    this.emitEvent({
+      type: "timeline",
+      provider: CODEX_PROVIDER,
+      item: mapCodexStreamRetryToStatusLine(retry, outcome),
+    });
+  }
+
  private resetTurnTrackingState(): void {
    this.latestPlanResult = null;
    this.emittedItemStartedIds.clear();
@@ -14,7 +14,7 @@ interface FakeSubAgentActivity {
  callId: string;
  threadId: string;
  agentPath: string;
-  kind: "started" | "interacted" | "interrupted";
+  kind: "started" | "interacted" | "interrupted" | "completed";
  parentThreadId?: string;
 }
 interface FakeLegacyCommand {
@@ -79,6 +79,14 @@ export interface FakeCodexAppServer {
    status?: "completed" | "failed" | "interrupted";
    error?: { message: string } | null;
  }): void;
+  reportsTurnError(params: {
+    threadId: string;
+    turnId: string;
+    message: string;
+    additionalDetails?: string;
+    willRetry: boolean;
+  }): void;
+  warns(params: { threadId: string | null; message: string }): void;
  startsSubAgent(params: {
    callId: string;
    threadId: string;
@@ -437,6 +445,21 @@ export function createFakeCodexAppServer(
        })}\n`,
      );
    },
+    reportsTurnError(params) {
+      writeNotification("error", {
+        threadId: params.threadId,
+        turnId: params.turnId,
+        willRetry: params.willRetry,
+        error: {
+          message: params.message,
+          codexErrorInfo: { responseStreamDisconnected: { httpStatusCode: null } },
+          additionalDetails: params.additionalDetails ?? null,
+        },
+      });
+    },
+    warns(params) {
+      writeNotification("warning", { threadId: params.threadId, message: params.message });
+    },
    startsSubAgent(params) {
      writeSubAgentActivity("item/completed", { ...params, kind: "started" });
    },