ci: re-run after a macOS file-observer flake

server-tests (macos-14, file observation) failed on file-observer/index.test.ts 'classifies a removed file as deleted', which this branch does not touch.

packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts

@@ -4,7 +4,11 @@ import { PassThrough } from "node:stream";
 import pino from "pino";
 import { describe, expect, test } from "vitest";
 
-import { JsonlRpcProcess, type JsonlRpcExit } from "./jsonl-rpc-process.js";
+import {
+  JsonlRpcProcess,
+  JsonlRpcRequestRejectedError,
+  type JsonlRpcExit,
+} from "./jsonl-rpc-process.js";
 
 const CHILD_SOURCE = String.raw`
 const readline = require("node:readline");
@@ -154,13 +158,13 @@ describe("JsonlRpcProcess", () => {
     }
   });
 
-  test("rejects unsuccessful responses", async () => {
+  test("rejects unsuccessful responses as rejected requests", async () => {
     const transport = startProcess();
 
     try {
-      await expect(transport.request({ type: "fail" })).rejects.toThrow(
-        "child rejected the request",
-      );
+      const request = transport.request({ type: "fail" });
+      await expect(request).rejects.toThrow("child rejected the request");
+      await expect(request).rejects.toBeInstanceOf(JsonlRpcRequestRejectedError);
     } finally {
       await transport.close();
     }
@@ -173,9 +177,12 @@ describe("JsonlRpcProcess", () => {
     try {
       child.stderr.write("still waiting");
 
-      await expect(transport.request({ type: "hang" })).rejects.toThrow(
+      const request = transport.request({ type: "hang" });
+      await expect(request).rejects.toThrow(
         /JSONL RPC request timed out phase=hang elapsedMs=\d+ timeoutMs=50\nstill waiting/,
       );
+      // A timed-out request may still be carried out, so it is not a rejection.
+      await expect(request).rejects.not.toBeInstanceOf(JsonlRpcRequestRejectedError);
     } finally {
       await transport.close();
     }

packages/server/src/server/agent/providers/jsonl-rpc-process.ts

@@ -40,6 +40,9 @@ interface PendingRequest {
   timer: NodeJS.Timeout | null;
 }
 
+/** The process answered the request with a failure, so it did not carry it out. */
+export class JsonlRpcRequestRejectedError extends Error {}
+
 export interface JsonlRpcExit {
   code: number | null;
   signal: NodeJS.Signals | null;
@@ -283,7 +286,7 @@ export class JsonlRpcProcess {
     this.pending.delete(response.id);
     if (!response.success) {
       pending.reject(
-        new Error(
+        new JsonlRpcRequestRejectedError(
           response.error ?? `${this.diagnosticName} ${response.command ?? "request"} failed`,
         ),
       );

packages/server/src/server/agent/providers/omp/agent.test.ts

@@ -8,6 +8,7 @@ import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js";
 import type { OmpUsagePollScheduler } from "./usage-poller.js";
 import { resolveOmpProviderParams } from "./provider-config.js";
 import { OmpHarness } from "./test-utils/omp-harness.js";
+import { JsonlRpcRequestRejectedError } from "../jsonl-rpc-process.js";
 
 const TURN_LIFECYCLE_EVENTS = new Set<AgentStreamEvent["type"]>([
   "turn_started",
@@ -740,6 +741,221 @@ describe("OMP agent client and session", () => {
     ]);
   });
 
+  test("a replacement prompt keeps its client correlation when OMP ends the aborted turn late", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    await omp.requireStartTurnFromClient("first prompt", "client-a");
+    const runtime = omp.runtime();
+    runtime.beginTurn();
+    runtime.acceptPrompt("first prompt", "omp-user-a");
+    runtime.streamAssistantText("working on it", "omp-assistant-a");
+    await omp.waitForUserMessages(1);
+
+    // OMP takes no steer, so Paseo interrupts and replaces the foreground turn.
+    await omp.interrupt();
+    await omp.requireStartTurnFromClient("second prompt", "client-b");
+    runtime.beginTurn();
+
+    // OMP flushes the aborted turn's terminal event after the replacement started.
+    runtime.emit({
+      type: "agent_end",
+      messages: [{ role: "assistant", content: [], responseId: "omp-assistant-a" }],
+    });
+    await omp.waitForEvent("turn_completed");
+
+    runtime.acceptPrompt("second prompt", "omp-user-b");
+    runtime.streamAssistantText("second answer", "omp-assistant-b");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(2);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      {
+        type: "user_message",
+        text: "first prompt",
+        messageId: "omp-user-a",
+        clientMessageId: "client-a",
+      },
+      {
+        type: "user_message",
+        text: "second prompt",
+        messageId: "omp-user-b",
+        clientMessageId: "client-b",
+      },
+    ]);
+  });
+
+  test("a prompt echoed after a hidden custom message completes the turn keeps its client correlation", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    await omp.requireStartTurnFromClient("do the thing", "client-a");
+    const runtime = omp.runtime();
+    runtime.beginTurn();
+    runtime.emit({
+      type: "message_end",
+      message: { role: "custom", content: "xd://inventory", display: false },
+    });
+    await omp.waitForEvent("turn_completed");
+
+    runtime.acceptPrompt("do the thing", "omp-user-a");
+    runtime.streamAssistantText("done", "omp-assistant-a");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(1);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      {
+        type: "user_message",
+        text: "do the thing",
+        messageId: "omp-user-a",
+        clientMessageId: "client-a",
+      },
+    ]);
+  });
+
+  test("a re-emitted echo does not consume a later submission's client correlation", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    await omp.requireStartTurnFromClient("again", "client-a");
+    const runtime = omp.runtime();
+    runtime.beginTurn();
+    runtime.acceptPrompt("again", "omp-user-a");
+    runtime.streamAssistantText("first answer", "omp-assistant-a");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(1);
+
+    await omp.requireStartTurnFromClient("again", "client-b");
+    runtime.beginTurn();
+    // OMP re-sends the first turn's entry before echoing the new submission.
+    runtime.acceptPrompt("again", "omp-user-a");
+    runtime.acceptPrompt("again", "omp-user-b");
+    runtime.streamAssistantText("second answer", "omp-assistant-b");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(2);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      { type: "user_message", text: "again", messageId: "omp-user-a", clientMessageId: "client-a" },
+      { type: "user_message", text: "again", messageId: "omp-user-b", clientMessageId: "client-b" },
+    ]);
+  });
+
+  test("a prompt OMP echoes after its dispatch failed keeps its client correlation", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    const runtime = omp.runtime();
+    // The prompt reached OMP, but Paseo's request failed before it was answered.
+    runtime.promptError = new Error("OMP prompt request timed out");
+    await omp.requireStartTurnFromClient("slow prompt", "client-a");
+    await omp.waitForEvent("turn_failed");
+    runtime.promptError = null;
+
+    // OMP accepted it anyway and echoes it once it gets to the turn.
+    runtime.beginTurn();
+    runtime.acceptPrompt("slow prompt", "omp-user-a");
+    runtime.streamAssistantText("late answer", "omp-assistant-a");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(1);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      {
+        type: "user_message",
+        text: "slow prompt",
+        messageId: "omp-user-a",
+        clientMessageId: "client-a",
+      },
+    ]);
+  });
+
+  test("a timed-out prompt OMP echoes late keeps its own correlation when the same text is resubmitted", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    const runtime = omp.runtime();
+    // OMP recorded the prompt, but Paseo's request timed out before it was answered.
+    runtime.promptError = new Error("OMP prompt request timed out");
+    await omp.requireStartTurnFromClient("retry me", "client-a");
+    await omp.waitForEvent("turn_failed");
+    runtime.promptError = null;
+
+    await omp.requireStartTurnFromClient("retry me", "client-b");
+    // OMP echoes both prompts in the order it received them.
+    runtime.beginTurn();
+    runtime.acceptPrompt("retry me", "omp-user-a");
+    runtime.acceptPrompt("retry me", "omp-user-b");
+    runtime.streamAssistantText("done", "omp-assistant-b");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(2);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      {
+        type: "user_message",
+        text: "retry me",
+        messageId: "omp-user-a",
+        clientMessageId: "client-a",
+      },
+      {
+        type: "user_message",
+        text: "retry me",
+        messageId: "omp-user-b",
+        clientMessageId: "client-b",
+      },
+    ]);
+  });
+
+  test("a prompt OMP rejects leaves no correlation for a later submission", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    const runtime = omp.runtime();
+    runtime.promptError = new JsonlRpcRequestRejectedError("OMP refused the prompt");
+    await omp.requireStartTurnFromClient("retry me", "client-a");
+    await omp.waitForEvent("turn_failed");
+    runtime.promptError = null;
+
+    await omp.requireStartTurnFromClient("retry me", "client-b");
+    runtime.beginTurn();
+    runtime.acceptPrompt("retry me", "omp-user-b");
+    runtime.streamAssistantText("done", "omp-assistant-b");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(1);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      {
+        type: "user_message",
+        text: "retry me",
+        messageId: "omp-user-b",
+        clientMessageId: "client-b",
+      },
+    ]);
+  });
+
+  test("the same prompt text on two turns correlates to its own submission", async () => {
+    const omp = new OmpHarness();
+    await omp.start();
+
+    await omp.requireStartTurnFromClient("again", "client-a");
+    const runtime = omp.runtime();
+    runtime.beginTurn();
+    runtime.acceptPrompt("again", "omp-user-a");
+    runtime.streamAssistantText("first answer", "omp-assistant-a");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(1);
+
+    await omp.requireStartTurnFromClient("again", "client-b");
+    runtime.beginTurn();
+    runtime.acceptPrompt("again", "omp-user-b");
+    runtime.streamAssistantText("second answer", "omp-assistant-b");
+    runtime.finishTurn();
+    await omp.waitForUserMessages(2);
+
+    expect(omp.timeline().filter((item) => item.type === "user_message")).toEqual([
+      { type: "user_message", text: "again", messageId: "omp-user-a", clientMessageId: "client-a" },
+      { type: "user_message", text: "again", messageId: "omp-user-b", clientMessageId: "client-b" },
+    ]);
+  });
+
   test("re-emitted user message_end frames dedupe by native entry id", async () => {
     const omp = new OmpHarness();
     await omp.start();

packages/server/src/server/agent/providers/omp/agent.ts

@@ -72,6 +72,7 @@ import { getUserMessageText } from "./message-history.js";
 import { mapOmpSystemNoticeToNotification } from "./system-notice.js";
 import { materializeProviderImage } from "../provider-image-output.js";
 import { OmpCliRuntime } from "./cli-runtime.js";
+import { JsonlRpcRequestRejectedError } from "../jsonl-rpc-process.js";
 import { listOmpImportableSessions, readOmpImportSessionConfig } from "./session-descriptor.js";
 import type { OmpRuntime, OmpRuntimeSession, OmpStartSessionInput } from "./runtime.js";
 import type {
@@ -322,6 +323,9 @@ function renderTextOnlyImageHint(image: { data: string; mimeType: string }): str
   }
 }
 
+// Bounds the dispatched prompts whose client correlation is still unclaimed.
+const MAX_PENDING_CLIENT_CORRELATIONS = 16;
+
 function convertPromptInput(
   prompt: AgentPromptInput,
   options: { model: OmpModel | null | undefined },
@@ -855,7 +859,12 @@ export class OmpAgentSession implements AgentSession {
   private activeAskUserDialog: ActiveAskUserDialog | null = null;
   private pendingCombinedAskUserResponse: PendingCombinedAskUserResponse | null = null;
   private activeTurnId: string | null = null;
-  private activeClientMessageId: string | null = null;
+  // A dispatched prompt owns its client correlation, not the turn that carried
+  // it: OMP echoes the prompt as a user message that can land after Paseo has
+  // already completed the turn (a replacement turn, or a turn completed by a
+  // custom message). Correlations are consumed by the echo they belong to, so
+  // the same text submitted on different turns stays separate.
+  private readonly pendingClientCorrelations: Array<{ text: string; clientMessageId: string }> = [];
   private activeAssistantMessageId: string | null = null;
   private activeTurnTerminalAssistantMessage: OmpAgentMessage | null = null;
   private activeTurnStarted = false;
@@ -962,9 +971,11 @@ export class OmpAgentSession implements AgentSession {
 
     const payload = convertPromptInput(prompt, { model: this.state.model });
     const turnId = randomUUID();
+    if (options?.clientMessageId) {
+      this.rememberClientCorrelation(payload.text, options.clientMessageId);
+    }
     this.live = true;
     this.activeTurnId = turnId;
-    this.activeClientMessageId = options?.clientMessageId ?? null;
     this.activeAssistantMessageId = null;
     this.activeTurnTerminalAssistantMessage = null;
     this.activeTurnStarted = false;
@@ -994,12 +1005,16 @@ export class OmpAgentSession implements AgentSession {
           return;
         }
       } catch (error) {
+        if (error instanceof JsonlRpcRequestRejectedError) {
+          // OMP refused the prompt, so no echo will ever claim it. Any other
+          // failure (a timeout) leaves the prompt with OMP, which still echoes it.
+          this.forgetClientCorrelation(options?.clientMessageId);
+        }
         if (this.activeTurnId !== turnId) {
           return;
         }
         this.usagePoller.stopTurn();
         this.activeTurnId = null;
-        this.activeClientMessageId = null;
         this.activeTurnStarted = false;
         this.activeTurnHasUserMessage = false;
         this.activeAssistantMessageId = null;
@@ -1132,7 +1147,6 @@ export class OmpAgentSession implements AgentSession {
       this.terminalizeActiveWork();
       this.usagePoller.stopTurn();
       this.activeTurnId = null;
-      this.activeClientMessageId = null;
       this.activeTurnStarted = false;
       this.activeTurnHasUserMessage = false;
       this.activeAssistantMessageId = null;
@@ -1293,6 +1307,36 @@ export class OmpAgentSession implements AgentSession {
     return this.activeTurnId ?? undefined;
   }
 
+  private rememberClientCorrelation(text: string, clientMessageId: string): void {
+    this.pendingClientCorrelations.push({ text, clientMessageId });
+    // OMP never echoes some prompts (a local-only command, a session that dies
+    // mid-turn). Keep the queue bounded so those cannot accumulate.
+    if (this.pendingClientCorrelations.length > MAX_PENDING_CLIENT_CORRELATIONS) {
+      this.pendingClientCorrelations.shift();
+    }
+  }
+
+  private forgetClientCorrelation(clientMessageId: string | undefined): void {
+    const index = this.pendingClientCorrelations.findIndex(
+      (entry) => entry.clientMessageId === clientMessageId,
+    );
+    if (index !== -1) {
+      this.pendingClientCorrelations.splice(index, 1);
+    }
+  }
+
+  // OMP echoes prompts in the order Paseo sent them, and a prompt OMP refused
+  // is forgotten on dispatch, so the oldest unclaimed correlation with the
+  // echoed text belongs to this echo.
+  private takeClientCorrelation(text: string): string | null {
+    const index = this.pendingClientCorrelations.findIndex((entry) => entry.text === text);
+    if (index === -1) {
+      return null;
+    }
+    const [entry] = this.pendingClientCorrelations.splice(index, 1);
+    return entry?.clientMessageId ?? null;
+  }
+
   private scheduleNoTurnPromptCompletion(turnId: string): void {
     this.cancelNoTurnPromptCompletion();
     const abort = new AbortController();
@@ -1346,6 +1390,7 @@ export class OmpAgentSession implements AgentSession {
     const outputs = this.pendingNoTurnOutputs.filter((output) => output.turnId === turnId);
     this.clearNoTurnBuffers();
     if (promptText) {
+      const clientMessageId = this.takeClientCorrelation(promptText);
       this.emit({
         type: "timeline",
         provider: this.provider,
@@ -1353,7 +1398,7 @@ export class OmpAgentSession implements AgentSession {
         item: {
           type: "user_message",
           text: promptText,
-          ...(this.activeClientMessageId ? { clientMessageId: this.activeClientMessageId } : {}),
+          ...(clientMessageId ? { clientMessageId } : {}),
         },
       });
     }
@@ -1791,7 +1836,6 @@ export class OmpAgentSession implements AgentSession {
     }
     const turnId = this.activeTurnId;
     this.activeTurnId = null;
-    this.activeClientMessageId = null;
     this.activeTurnStarted = false;
     this.activeTurnHasUserMessage = false;
     this.activeTurnTerminalAssistantMessage = null;
@@ -2048,7 +2092,6 @@ export class OmpAgentSession implements AgentSession {
     }
     const nativeMessage = event.message as OmpAgentMessage & { id?: unknown; entryId?: unknown };
     const messageId = readNativeMessageId(nativeMessage);
-    const clientMessageId = this.activeClientMessageId;
     const emitUserMessage = (resolvedMessageId?: string): void => {
       if (resolvedMessageId) {
         // OMP re-emits user message_end frames for entries it has already
@@ -2059,6 +2102,9 @@ export class OmpAgentSession implements AgentSession {
         }
         this.emittedUserMessageIds.add(resolvedMessageId);
       }
+      // Claimed only once a row is really emitted, so a re-emitted frame cannot
+      // consume the correlation of a later submission with the same text.
+      const clientMessageId = this.takeClientCorrelation(text);
       this.emit({
         type: "timeline",
         provider: this.provider,
@@ -2132,7 +2178,6 @@ export class OmpAgentSession implements AgentSession {
 
   private completeTurn(turnId: string | undefined, messages: OmpAgentMessage[]): void {
     this.activeTurnId = null;
-    this.activeClientMessageId = null;
     this.activeAssistantMessageId = null;
     this.activeTurnTerminalAssistantMessage = null;
     this.activeTurnStarted = false;

packages/server/src/server/agent/providers/omp/runtime.ts

@@ -44,6 +44,7 @@ export interface OmpStartSessionInput {
 
 export interface OmpRuntimeSession {
   onEvent(callback: (event: OmpRuntimeEvent) => void): () => void;
+  // Rejects with JsonlRpcRequestRejectedError when OMP refused the prompt.
   prompt(
     message: string,
     images?: Array<{ type: "image"; data: string; mimeType: string }>,

packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts

@@ -133,6 +133,7 @@ export class FakeOmpSession implements OmpRuntimeSession {
   compactError: Error | null = null;
   emitCompactEnd = true;
   getStateError: Error | null = null;
+  promptError: Error | null = null;
   promptAck: OmpPromptAck = {};
   branchResponse: { text?: string; cancelled?: boolean } = { text: "" };
   branchMessages: Array<{ entryId: string; text: string }> = [];
@@ -182,6 +183,9 @@ export class FakeOmpSession implements OmpRuntimeSession {
     }
     this.prompts.push({ message, imageCount: images?.length ?? 0 });
     this.promptWaiters.shift()?.();
+    if (this.promptError) {
+      throw this.promptError;
+    }
     const heldPrompt = this.nextHeldPrompt;
     if (heldPrompt) {
       this.nextHeldPrompt = null;

packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts

@@ -466,6 +466,33 @@ export class OmpHarness {
     await promptStarted;
   }
 
+  // Drains the microtask queue until the session has reached the state the test
+  // is waiting on, so no test depends on how many async steps a path takes.
+  async waitUntil(condition: () => boolean, description: string): Promise<void> {
+    for (let index = 0; index < 100; index += 1) {
+      if (condition()) return;
+      await waitForImmediate();
+    }
+    throw new Error(`OMP harness timed out waiting for ${description}`);
+  }
+
+  async waitForUserMessages(count: number): Promise<void> {
+    await this.waitUntil(
+      () => this.timeline().filter((item) => item.type === "user_message").length >= count,
+      `${count} user message(s)`,
+    );
+  }
+
+  async waitForEvent(type: AgentStreamEvent["type"]): Promise<void> {
+    await this.waitUntil(() => this.eventTypes().includes(type), `a ${type} event`);
+  }
+
+  async requireStartTurnFromClient(message: string, clientMessageId: string): Promise<void> {
+    const promptStarted = this.omp.latestSession().nextPrompt();
+    await this.requireSession().startTurn(message, { clientMessageId });
+    await promptStarted;
+  }
+
   async interrupt(): Promise<void> {
     await this.requireSession().interrupt();
   }

packages/server/src/server/workspace-git-service.observation.test.ts

@@ -1006,7 +1006,13 @@ describe("WorkspaceGitService checkout observation", () => {
       branchMergeRef: null,
       upstreamStatus: null,
     }));
-    const service = createService(watcher, { getCheckoutSnapshotFacts });
+    // A remote URL starts the background fetch; keep it off the real git binary.
+    const runGitFetch = vi.fn(async () => ({
+      changes: [],
+      nonRemoteRefsChanged: false,
+      error: null,
+    }));
+    const service = createService(watcher, { getCheckoutSnapshotFacts, runGitFetch });
     const subscription = service.registerWorkspace({ cwd: REPO_CWD }, vi.fn());
     await vi.waitFor(() => {
       expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(1);