fix(pi): avoid delivery warnings for consumed image steers

Pi retains empty text entries in its steering queue after consuming image-only input. Require an unconsumed submission before reporting those entries as removed, and model that native behavior in the fake.

docs/android.md

@@ -174,7 +174,7 @@ REACT_NATIVE_PACKAGER_HOSTNAME=localhost \
 
 This is the Android counterpart of the iOS local-simulator flow in [development.md](development.md): on iOS the simulator shares the Mac's loopback so `localhost:<port>` works directly; on Android you need `10.0.2.2` or `adb reverse`.
 
-## Inverted timeline selection
+## Inverted timeline input
 
 Android focus and selection visibility requests must not reposition inverted timelines. The
 `modules/paseo-scroll` package keeps React Native's scroll manager interface and returns zero for
@@ -183,6 +183,10 @@ Native's Android list uses `scale: -1`, which decomposes into a 180-degree rotat
 `scaleY`; checking the scale property alone misses it. Dragging and explicit scroll
 commands still work; non-inverted scroll views keep Android's default behavior.
 
+Mouse wheel axes also bypass the inversion transform. The same module reverses the vertical
+wheel axis for inverted scroll views before delegating to Android, preserving native scroll
+bounds and React Native's scroll events and disabled-scroll behavior.
+
 Register this package before React Native's core package through its Expo config plugin. Normal
 Android builds use the prebuilt `react-android` library, so patching Java under `node_modules` does
 not change the shipped scroll view. Keep this behavior in the app's compiled native module.

docs/custom-providers.md

@@ -777,6 +777,8 @@ The built-in `claude` provider appends concrete model IDs from `~/.claude/settin
 
 This lets users who already configured Claude Code for Bedrock, OpenRouter, ollama, Z.AI, or another Anthropic-compatible gateway select the exact model ID in Paseo. Explicit model IDs are passed unchanged to Claude Code, even when the same string is a compatibility alias for a built-in model. When `agents.providers.claude.models` is set it **replaces** both the hardcoded first-party Claude list and any settings.json-discovered entries; use `agents.providers.claude.additionalModels` to keep the first-party list and append curated entries on top.
 
+Paseo's effort selector does not inherit `effortLevel` from Claude's settings.json. Select an effort in Paseo to choose a specific value.
+
 ### Gotcha: `extends: "claude"` with third-party endpoints
 
 When a custom provider extends `"claude"` but points `ANTHROPIC_BASE_URL` at a non-Anthropic API (Z.AI, Alibaba/Qwen, proxies), the Claude Agent SDK may try to use Anthropic-only server-side tools like `WebSearch`. Third-party APIs don't support these tools, causing errors.

packages/app/modules/paseo-scroll/android/src/androidTest/java/sh/paseo/scroll/ScrollSelectionTest.kt

@@ -4,11 +4,13 @@ import android.graphics.Rect
 import android.text.Selection
 import android.text.Spannable
 import android.view.View
+import android.view.InputDevice
+import android.view.MotionEvent
+import android.os.SystemClock
 import android.widget.FrameLayout
 import android.widget.TextView
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import androidx.test.platform.app.InstrumentationRegistry
-import com.facebook.react.bridge.BridgeReactContext
 import com.facebook.react.bridge.JavaOnlyArray
 import com.facebook.react.bridge.JavaOnlyMap
 import com.facebook.react.soloader.OpenSourceMergedSoMapping
@@ -24,6 +26,40 @@ import org.junit.runner.RunWith
 
 @RunWith(AndroidJUnit4::class)
 class ScrollSelectionTest {
+  @Test fun mouseWheelMovesInvertedViewportInVisualDirection() = withScroll(inverted = true) { scroll, _ ->
+    val before = scroll.scrollY
+    withWheel(1f) { event ->
+      assertTrue(scroll.dispatchGenericMotionEvent(event))
+      assertTrue("Wheel up must move toward older messages: $before -> ${scroll.scrollY}", scroll.scrollY > before)
+      assertEquals("The caller's event must not change", 1f, event.getAxisValue(MotionEvent.AXIS_VSCROLL))
+    }
+    val afterUp = scroll.scrollY
+    withWheel(-1f) { event ->
+      assertTrue(scroll.dispatchGenericMotionEvent(event))
+      assertTrue("Wheel down must move toward newer messages: $afterUp -> ${scroll.scrollY}", scroll.scrollY < afterUp)
+    }
+  }
+
+  @Test fun mouseWheelKeepsOrdinaryScrollDirection() = withScroll(inverted = false) { scroll, _ ->
+    val before = scroll.scrollY
+    withWheel(1f) { event ->
+      assertTrue(scroll.dispatchGenericMotionEvent(event))
+      assertTrue(scroll.scrollY < before)
+    }
+    val afterUp = scroll.scrollY
+    withWheel(-1f) { event ->
+      assertTrue(scroll.dispatchGenericMotionEvent(event))
+      assertTrue(scroll.scrollY > afterUp)
+    }
+  }
+
+  @Test fun mouseWheelRespectsDisabledScrolling() = withScroll(inverted = true) { scroll, _ ->
+    scroll.setScrollEnabled(false)
+    val before = scroll.scrollY
+    withWheel(1f) { scroll.dispatchGenericMotionEvent(it) }
+    assertEquals(before, scroll.scrollY)
+  }
+
   @Test fun focusingSelectableTextKeepsInvertedViewportStill() = withScroll(inverted = true) { scroll, text ->
     val before = scroll.scrollY
     assertTrue(text.requestFocus())
@@ -70,7 +106,7 @@ class ScrollSelectionTest {
       val appContext = instrumentation.targetContext
       DisplayMetricsHolder.initDisplayMetricsIfNotInitialized(appContext)
       SoLoader.init(appContext, OpenSourceMergedSoMapping)
-      val reactContext = BridgeReactContext(appContext)
+      val reactContext = ScrollTestContext(appContext)
       val context = ThemedReactContext(reactContext, appContext)
       val manager = PaseoScrollPackage().createViewManager(reactContext, ReactScrollViewManager.REACT_CLASS) as ReactScrollViewManager
       val scroll = manager.createViewInstance(context)
@@ -98,4 +134,26 @@ class ScrollSelectionTest {
   }
 
   private fun exact(size: Int) = View.MeasureSpec.makeMeasureSpec(size, View.MeasureSpec.EXACTLY)
+
+  private fun withWheel(axis: Float, check: (MotionEvent) -> Unit) {
+    val time = SystemClock.uptimeMillis()
+    val properties = MotionEvent.PointerProperties().apply {
+      id = 0
+      toolType = MotionEvent.TOOL_TYPE_MOUSE
+    }
+    val coordinates = MotionEvent.PointerCoords().apply {
+      x = 150f
+      y = 150f
+      setAxisValue(MotionEvent.AXIS_VSCROLL, axis)
+    }
+    val event = MotionEvent.obtain(
+      time, time, MotionEvent.ACTION_SCROLL, 1, arrayOf(properties), arrayOf(coordinates),
+      0, 0, 1f, 1f, 0, 0, InputDevice.SOURCE_MOUSE, 0,
+    )
+    try {
+      check(event)
+    } finally {
+      event.recycle()
+    }
+  }
 }

packages/app/modules/paseo-scroll/android/src/androidTest/java/sh/paseo/scroll/ScrollTestContext.java

@@ -0,0 +1,19 @@
+package sh.paseo.scroll;
+
+import android.content.Context;
+import com.facebook.react.bridge.BridgeReactContext;
+import com.facebook.react.uimanager.events.BlackHoleEventDispatcher;
+import com.facebook.react.uimanager.events.EventDispatcher;
+import com.facebook.react.uimanager.events.EventDispatcherProvider;
+
+/** Native-only tests use RN's dispatcher for a context without a JS instance. */
+final class ScrollTestContext extends BridgeReactContext implements EventDispatcherProvider {
+  ScrollTestContext(Context context) {
+    super(context);
+  }
+
+  @Override
+  public EventDispatcher getEventDispatcher() {
+    return BlackHoleEventDispatcher.INSTANCE;
+  }
+}

packages/app/modules/paseo-scroll/android/src/main/java/sh/paseo/scroll/PaseoScrollPackage.kt

@@ -2,6 +2,8 @@ package sh.paseo.scroll
 
 import android.graphics.Matrix
 import android.graphics.Rect
+import android.view.InputDevice
+import android.view.MotionEvent
 import com.facebook.react.ReactPackage
 import com.facebook.react.ViewManagerOnDemandReactPackage
 import com.facebook.react.bridge.NativeModule
@@ -30,13 +32,47 @@ private class PaseoScrollViewManager : ReactScrollViewManager() {
 private class PaseoScrollView(context: ThemedReactContext) : ReactScrollView(context) {
   private val transformValues = FloatArray(9)
 
+  private fun isVerticallyInverted(): Boolean {
+    // Android FlatList uses scale: -1, which RN decomposes into a 180-degree
+    // rotation with positive scaleY. Read the resulting transform, not scaleY.
+    matrix.getValues(transformValues)
+    return transformValues[Matrix.MSCALE_Y] < 0
+  }
+
+  override fun onGenericMotionEvent(event: MotionEvent): Boolean {
+    if (!isVerticallyInverted() || event.actionMasked != MotionEvent.ACTION_SCROLL ||
+        !event.isFromSource(InputDevice.SOURCE_CLASS_POINTER)) {
+      return super.onGenericMotionEvent(event)
+    }
+    if (!scrollEnabled) return false
+
+    // Android applies wheel axes directly to scrollY; the inversion transform only
+    // transforms pointer coordinates. Reverse the vertical axis before delegating so
+    // platform scroll bounds and scroll events still apply.
+    val properties = Array(event.pointerCount) { index ->
+      MotionEvent.PointerProperties().also { event.getPointerProperties(index, it) }
+    }
+    val coordinates = Array(event.pointerCount) { index ->
+      MotionEvent.PointerCoords().also {
+        event.getPointerCoords(index, it)
+        it.setAxisValue(MotionEvent.AXIS_VSCROLL, -it.getAxisValue(MotionEvent.AXIS_VSCROLL))
+      }
+    }
+    val correctedEvent = MotionEvent.obtain(
+      event.downTime, event.eventTime, event.action, event.pointerCount, properties, coordinates,
+      event.metaState, event.buttonState, event.xPrecision, event.yPrecision,
+      event.deviceId, event.edgeFlags, event.source, event.flags,
+    )
+    return try {
+      super.onGenericMotionEvent(correctedEvent)
+    } finally {
+      correctedEvent.recycle()
+    }
+  }
+
   // Both RN focus handling and Android selection visibility requests use this calculation.
   // Its coordinates ignore inversion. Keep focus/selection bookkeeping, but leave timeline
   // movement to user gestures and explicit scroll commands.
-  override fun computeScrollDeltaToGetChildRectOnScreen(rect: Rect): Int {
-    // RN's Android scale: -1 becomes a 180-degree rotation with positive scaleY.
-    matrix.getValues(transformValues)
-    if (transformValues[Matrix.MSCALE_Y] < 0) return 0
-    return super.computeScrollDeltaToGetChildRectOnScreen(rect)
-  }
+  override fun computeScrollDeltaToGetChildRectOnScreen(rect: Rect): Int =
+    if (isVerticallyInverted()) 0 else super.computeScrollDeltaToGetChildRectOnScreen(rect)
 }

packages/server/src/server/agent/agent-manager.test.ts

@@ -58,6 +58,9 @@ import type {
 import type { PaseoToolCatalog } from "./tools/types.js";
 import type { ProviderDefinition } from "./provider-registry.js";
 
+import { OpenCodeV2AgentClient } from "./providers/opencode/v2/agent.js";
+import { V2Harness } from "./providers/opencode/test-utils/v2-harness.js";
+
 const DESKTOP_OPEN_AGENT_TAB_LABEL = getOpenAgentTabLabel("desktop-client");
 const MOBILE_OPEN_AGENT_TAB_LABEL = getOpenAgentTabLabel("mobile-client");
 
@@ -670,6 +673,33 @@ async function startAndSteerThroughManager(
   return { manager, agentId: agent.id, workdir };
 }
 
+test("stops an OpenCode execution already active when its conversation is reopened", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-opencode-resume-"));
+  onTestFinished(() => rmSync(workdir, { recursive: true, force: true }));
+  const harness = new V2Harness();
+  harness.info.location.directory = workdir;
+  harness.startExecution();
+  const client = new OpenCodeV2AgentClient({ logger, runtime: harness.runtime });
+  const manager = new AgentManager({ clients: { opencode: client }, logger });
+  const agent = await manager.resumeAgentFromPersistence({
+    provider: "opencode",
+    sessionId: "session",
+    nativeHandle: "session",
+    metadata: { cwd: workdir },
+  });
+  try {
+    expect(await manager.cancelAgentRun(agent.id)).toEqual({ status: "settled" });
+    expect(harness.active).toBe(false);
+    for await (const _event of manager.streamAgent(agent.id, "continue after restart")) {
+      // Drain the next turn through the same interface used by callers.
+    }
+    expect(harness.prompts).toEqual(["continue after restart"]);
+  } finally {
+    await manager.closeAgent(agent.id);
+    await client.shutdown();
+  }
+});
+
 test("uses an injected timeline store without making it a production requirement", async () => {
   const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-store-"));
   const store = new RecordingTimelineStore();

packages/server/src/server/agent/providers/claude/models.test.ts

@@ -365,7 +365,21 @@ describe("ClaudeAgentClient.fetchCatalog", () => {
     );
   });
 
-  it("appends concrete models from Claude settings.json", async () => {
+  it("appends concrete models from Claude settings.json with thinking options", async () => {
+    const customModelThinkingOptions = [
+      { id: "low", label: "Low" },
+      { id: "medium", label: "Medium" },
+      { id: "high", label: "High", isDefault: true },
+      { id: "max", label: "Max" },
+    ];
+    const fableThinkingOptions = [
+      { id: "low", label: "Low" },
+      { id: "medium", label: "Medium" },
+      { id: "high", label: "High", isDefault: true },
+      { id: "xhigh", label: "Extra High" },
+      { id: "max", label: "Max" },
+      { id: CLAUDE_ULTRACODE_THINKING_OPTION_ID, label: "Ultra Code" },
+    ];
     const configDir = await createClaudeConfigDir({
       model: "us.anthropic.claude-opus-4-7[1m]",
       env: {
@@ -393,42 +407,49 @@ describe("ClaudeAgentClient.fetchCatalog", () => {
         id: "us.anthropic.claude-opus-4-7[1m]",
         label: "us.anthropic.claude-opus-4-7[1m]",
         description: "From Claude settings.json model",
+        thinkingOptions: customModelThinkingOptions,
       },
       {
         provider: "claude",
         id: "openrouter/anthropic/claude-sonnet-4.5",
         label: "openrouter/anthropic/claude-sonnet-4.5",
         description: "From Claude settings.json env.ANTHROPIC_MODEL",
+        thinkingOptions: customModelThinkingOptions,
       },
       {
         provider: "claude",
         id: "ollama/qwen3-coder",
         label: "ollama/qwen3-coder",
         description: "From Claude settings.json env.ANTHROPIC_SMALL_FAST_MODEL",
+        thinkingOptions: customModelThinkingOptions,
       },
       {
         provider: "claude",
         id: "claude-fable-5-1[1M]",
         label: "claude-fable-5-1[1M]",
         description: "From Claude settings.json env.ANTHROPIC_DEFAULT_FABLE_MODEL",
+        thinkingOptions: fableThinkingOptions,
       },
       {
         provider: "claude",
         id: "bedrock-opus-from-env",
         label: "bedrock-opus-from-env",
         description: "From Claude settings.json env.ANTHROPIC_DEFAULT_OPUS_MODEL",
+        thinkingOptions: customModelThinkingOptions,
       },
       {
         provider: "claude",
         id: "glm-5.1",
         label: "glm-5.1",
         description: "From Claude settings.json env.ANTHROPIC_DEFAULT_SONNET_MODEL",
+        thinkingOptions: customModelThinkingOptions,
       },
       {
         provider: "claude",
         id: "glm-5",
         label: "glm-5",
         description: "From Claude settings.json env.ANTHROPIC_DEFAULT_HAIKU_MODEL",
+        thinkingOptions: customModelThinkingOptions,
       },
     ]);
     expect(models.filter((model) => model.id === "claude-fable-5-1[1M]")).toHaveLength(1);

packages/server/src/server/agent/providers/claude/models.ts

@@ -83,7 +83,7 @@ export async function getClaudeModelsWithSettings(
     const manifestModel = getClaudeModels(claudeCodeVersion).find(
       (candidate) => candidate.id === model.id,
     );
-    models.push({ ...manifestModel, ...model });
+    models.push(resolveConfiguredClaudeModel({ ...manifestModel, ...model }));
   }
 
   return models;

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

@@ -193,7 +193,8 @@ export class JsonlRpcProcess {
    * Send a command whose goal an exited process has already met — stopping a turn,
    * clearing a queue. Resolves once the child has exited instead of rejecting,
    * because nothing is queued and nothing is running, which is what the caller
-   * asked for. Use `request` when the caller needs an answer from a live process.
+   * asked for. Returns the native reply when available, or undefined after exit.
+   * Use `request` when the caller requires an answer from a live process.
    *
    * Keyed on an observed exit rather than on `disposed`: `close()` disposes the
    * transport before termination completes, and a child that outlives SIGKILL can
@@ -202,12 +203,12 @@ export class JsonlRpcProcess {
   async requestStopWork(
     command: { type: string; [key: string]: unknown },
     timeoutMs?: number | null,
-  ): Promise<void> {
+  ): Promise<unknown> {
     if (this.exited) {
       return;
     }
     try {
-      await this.request(command, timeoutMs);
+      return await this.request(command, timeoutMs);
     } catch (error) {
       // A dying child breaks its stdin pipe a fraction before it emits `exit`, which
       // fails this request and starts a termination. Wait for that termination to reach

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

@@ -307,8 +307,8 @@ class OmpCliRuntimeSession implements OmpRuntimeSession {
     return this.process.request(OmpRpcCommandSchema.parse(command), timeoutMs);
   }
 
-  private requestStopWork(command: OmpRpcCommand): Promise<void> {
-    return this.process.requestStopWork(OmpRpcCommandSchema.parse(command));
+  private async requestStopWork(command: OmpRpcCommand): Promise<void> {
+    await this.process.requestStopWork(OmpRpcCommandSchema.parse(command));
   }
 
   private emit(event: OmpRuntimeEvent): void {

packages/server/src/server/agent/providers/opencode/v2/session.ts

@@ -242,6 +242,10 @@ export class OpenCodeV2Session implements AgentSession {
   }
   subscribe(callback: (event: AgentStreamEvent) => void) {
     this.listeners.add(callback);
+    // Reopening can adopt an execution during initialization, before its caller
+    // can subscribe. Deliver that live turn so it remains visible and interruptible.
+    if (this.turns.id)
+      callback({ type: "turn_started", provider: "opencode", turnId: this.turns.id });
     return () => {
       this.listeners.delete(callback);
     };

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

@@ -208,52 +208,6 @@ async function createSession(
   return { pi, session, events };
 }
 
-test("close preserves a foreground tool's terminal update before releasing Pi", async () => {
-  const { pi, session, events } = await createSession();
-  const native = pi.latestSession();
-  const stopped = Promise.withResolvers<void>();
-  native.abortCompletion = stopped.promise;
-  await session.startTurn("Run a foreground command");
-  native.emit({
-    type: "tool_execution_start",
-    toolCallId: "foreground-bash",
-    toolName: "bash",
-    args: { command: "sleep 120" },
-  });
-
-  const closing = session.close();
-  await flushTurnScheduling();
-  // Pi's abort acknowledgement follows its final tool result and durable transcript write.
-  native.emit({
-    type: "tool_execution_end",
-    toolCallId: "foreground-bash",
-    toolName: "bash",
-    result: { content: [{ type: "text", text: "Command aborted" }] },
-    isError: true,
-  });
-  stopped.resolve();
-  await closing;
-
-  expect(
-    events
-      .timelineItems()
-      .findLast((item) => item.type === "tool_call" && item.callId === "foreground-bash"),
-  ).toMatchObject({
-    callId: "foreground-bash",
-    status: "failed",
-  });
-});
-
-test("close releases Pi when stopping its active work fails", async () => {
-  const { pi, session } = await createSession();
-  const native = pi.latestSession();
-  await session.startTurn("Run a foreground command");
-  native.clearQueueError = new Error("Pi control plane is unavailable");
-
-  await expect(session.close()).resolves.toBeUndefined();
-  expect(native.closeCalls).toBe(1);
-});
-
 test("forwards launch-context env to the Pi process launch", async () => {
   const pi = new FakePi();
   const client = createClient(pi);
@@ -2536,6 +2490,322 @@ describe("PiRpcAgentSession", () => {
 });
 
 describe("PiRpcAgentSession steering", () => {
+  test.each([false, true])(
+    "does not warn for Pi's stale consumed image queue entry (native echo: %s)",
+    async (echoEntry) => {
+      const { pi, session, events } = await createSession();
+      const native = pi.latestSession();
+      native.setModelResult = {
+        provider: "openai",
+        id: "gpt-4o",
+        input: ["text", "image"],
+      };
+      await session.setModel("openai/gpt-4o");
+      const { turnId } = await session.startTurn("work");
+      await session.steerActiveTurn(
+        [{ type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" }],
+        { expectedTurnId: turnId },
+      );
+      if (echoEntry) {
+        native.finishSubmittedUserMessage({ id: "image-steer", parentId: null, text: "" });
+      } else {
+        native.emit({ type: "message_end", message: { role: "user", content: [] } });
+      }
+
+      await session.interrupt();
+
+      expect(events.timelineItems().filter((item) => item.type === "notification")).toEqual([]);
+    },
+  );
+
+  test("does not report a consumed steer rewritten by a Pi input extension", async () => {
+    const { pi, session, events } = await createSession();
+    pi.latestSession().steerTextTransform = (text) => text.replace("please ", "");
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("please continue", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    pi.latestSession().finishSubmittedUserMessage({
+      id: "native-transformed-steer",
+      parentId: null,
+      text: "continue",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems().filter((item) => item.type === "notification")).toEqual([]);
+  });
+
+  test("names the rewritten steer that Pi actually removed", async () => {
+    const { pi, session, events } = await createSession();
+    pi.latestSession().steerTextTransform = (text) => text.replace("please ", "");
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("please continue", { expectedTurnId: turnId });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toContainEqual({
+      type: "notification",
+      level: "warning",
+      message: "Not delivered (stopped): continue",
+    });
+  });
+
+  test("does not consume a queued steer when the outstanding foreground prompt has the same text", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("continue", {
+      clientMessageId: "client-prompt-1",
+    });
+    await session.steerActiveTurn("continue", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    pi.latestSession().emit({ type: "agent_start" });
+    pi.latestSession().emit({ type: "turn_start" });
+    pi.latestSession().finishSubmittedUserMessage({
+      id: "native-prompt-1",
+      parentId: null,
+      text: "continue",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "continue",
+        messageId: "native-prompt-1",
+        clientMessageId: "client-prompt-1",
+      },
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): continue",
+      },
+    ]);
+  });
+
+  test("keeps a consumed steer distinct from an identical foreground prompt", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("continue", { clientMessageId: "client-prompt-1" });
+    await session.steerActiveTurn("continue", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    const native = pi.latestSession();
+    native.finishSubmittedUserMessage({ id: "native-prompt-1", parentId: null, text: "continue" });
+    native.finishSubmittedUserMessage({
+      id: "native-steer-1",
+      parentId: "native-prompt-1",
+      text: "continue",
+    });
+    await session.interrupt();
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "continue",
+        messageId: "native-prompt-1",
+        clientMessageId: "client-prompt-1",
+      },
+      {
+        type: "user_message",
+        text: "continue",
+        messageId: "native-steer-1",
+        clientMessageId: "client-steer-1",
+      },
+    ]);
+  });
+
+  test.each([1, 2])(
+    "labels a stopped image-only steer containing %i image(s)",
+    async (imageCount) => {
+      const { pi, session, events } = await createSession();
+      const native = pi.latestSession();
+      native.setModelResult = {
+        provider: "openai",
+        id: "gpt-4o",
+        name: "GPT-4o",
+        input: ["text", "image"],
+      };
+      await session.setModel("openai/gpt-4o");
+      const { turnId } = await session.startTurn([
+        { type: "image", data: ONE_BY_ONE_PNG_BASE64, mimeType: "image/png" },
+      ]);
+      await session.steerActiveTurn(
+        Array.from({ length: imageCount }, () => ({
+          type: "image" as const,
+          data: ONE_BY_ONE_PNG_BASE64,
+          mimeType: "image/png",
+        })),
+        { expectedTurnId: turnId, clientMessageId: "client-steer-1" },
+      );
+      native.emit({ type: "message_end", message: { role: "user", content: [] } });
+      await session.interrupt();
+      expect(native.steerCalls).toEqual([{ message: "", imageCount }]);
+      expect(events.timelineItems()).toEqual([
+        {
+          type: "notification",
+          level: "warning",
+          message: `Not delivered (stopped): ${imageCount} ${imageCount === 1 ? "image" : "images"}`,
+        },
+      ]);
+    },
+  );
+
+  test("reports an unconsumed steer as not delivered when Stop clears the queue", async () => {
+    const { session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("Reply STEER_CHECK_7391", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toContainEqual({
+      type: "notification",
+      level: "warning",
+      message: "Not delivered (stopped): Reply STEER_CHECK_7391",
+    });
+  });
+
+  test("does not report a consumed steer before its native entry ID arrives", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("already consumed", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    pi.latestSession().emit({
+      type: "message_end",
+      message: { role: "user", content: "already consumed" },
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([]);
+  });
+
+  test("tracks native consumption before the steer RPC is acknowledged", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    const native = pi.latestSession();
+    native.steer = async (message) => {
+      native.finishSubmittedUserMessage({ id: "native-entry-1", parentId: null, text: message });
+    };
+
+    await expect(
+      session.steerActiveTurn("already consumed", {
+        expectedTurnId: turnId,
+        clientMessageId: "client-steer-1",
+      }),
+    ).resolves.toEqual({ status: "accepted" });
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "already consumed",
+        messageId: "native-entry-1",
+        clientMessageId: "client-steer-1",
+      },
+    ]);
+  });
+
+  test("keeps identical steers distinct from native message IDs when one is consumed", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("same text", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    await session.steerActiveTurn("same text", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-2",
+    });
+    pi.latestSession().finishSubmittedUserMessage({
+      id: "native-entry-1",
+      parentId: null,
+      text: "same text",
+    });
+
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "user_message",
+        text: "same text",
+        messageId: "native-entry-1",
+        clientMessageId: "client-steer-1",
+      },
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): same text",
+      },
+    ]);
+  });
+
+  test("retains steer tracking when clear_queue fails so Stop can be retried", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("queued", {
+      expectedTurnId: turnId,
+      clientMessageId: "client-steer-1",
+    });
+    const native = pi.latestSession();
+    native.clearQueueError = new Error("Queue clear failed");
+
+    await expect(session.interrupt()).rejects.toThrow("Queue clear failed");
+    expect(native.controlRequests).toEqual(["clear_queue"]);
+    expect(events.timelineItems()).toEqual([]);
+
+    native.clearQueueError = null;
+    await session.interrupt();
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): queued",
+      },
+    ]);
+  });
+
+  test("reports an acknowledged queue removal even when abort subsequently fails", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("queued", { expectedTurnId: turnId });
+    const native = pi.latestSession();
+    native.abort = async () => {
+      throw new Error("Abort failed");
+    };
+
+    await expect(session.interrupt()).rejects.toThrow("Abort failed");
+    native.abort = async () => {};
+    await session.interrupt();
+
+    expect(events.timelineItems()).toEqual([
+      {
+        type: "notification",
+        level: "warning",
+        message: "Not delivered (stopped): queued",
+      },
+    ]);
+  });
+
+  test("does not claim queue removal when an older Pi lacks clear_queue", async () => {
+    const { pi, session, events } = await createSession();
+    const { turnId } = await session.startTurn("work");
+    await session.steerActiveTurn("queued", { expectedTurnId: turnId });
+    pi.latestSession().clearQueueError = new Error("Unknown command: clear_queue");
+
+    await session.interrupt();
+
+    expect(pi.latestSession().controlRequests).toEqual(["clear_queue", "abort"]);
+    expect(events.timelineItems()).toEqual([]);
+  });
+
   test("steers the active Pi turn and correlates the echoed user message", async () => {
     const { pi, session, events } = await createSession();
     const fakeSession = pi.latestSession();

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

@@ -83,7 +83,12 @@ import {
   listPiImportableSessions,
   readPiImportSessionConfig,
 } from "./session-descriptor.js";
-import type { PiRuntime, PiRuntimeSession, PiStartSessionInput } from "./runtime.js";
+import type {
+  PiClearedQueue,
+  PiRuntime,
+  PiRuntimeSession,
+  PiStartSessionInput,
+} from "./runtime.js";
 import type {
   PiAgentSessionEvent,
   PiAgentMessage,
@@ -232,7 +237,14 @@ interface StartTurnResult {
 
 interface PiPendingSteerSubmission {
   text: string;
+  label: string;
   clientMessageId: string | null;
+  consumed: boolean;
+}
+
+interface PiPendingForegroundSubmission {
+  text: string;
+  consumed: boolean;
 }
 
 // COMPAT(piSteerFallback): added in v0.5.0, remove after 2027-03-01 once the pi floor
@@ -1210,6 +1222,7 @@ export class PiRpcAgentSession implements AgentSession {
   private activePromptRequestId: string | null = null;
   private readonly pendingPromptResults = new Map<string, boolean>();
   private readonly pendingSteerSubmissions: PiPendingSteerSubmission[] = [];
+  private pendingForegroundSubmission: PiPendingForegroundSubmission | null = null;
   currentLeafOverrideId: string | null | undefined;
   private readonly treeUserEntriesById = new Map<string, PiCapturedEntry>();
   private readonly pendingExtensionResults = new Map<string, PendingExtensionResult>();
@@ -1319,10 +1332,12 @@ export class PiRpcAgentSession implements AgentSession {
     this.pendingSettledMessages = null;
     this.activePromptRequestId = null;
     this.pendingSteerSubmissions.length = 0;
+    this.pendingForegroundSubmission = null;
     this.clearNoTurnBuffers();
     const payload = await convertPromptInput(prompt, { model: this.state.model });
     if (this.closed) throw new Error("Pi session is closed");
     if (this.activeTurnId !== turnId) return { turnId };
+    this.pendingForegroundSubmission = { text: payload.text, consumed: false };
     this.activeNoTurnPromptText = payload.text;
     const shouldProbeForNoTurnPrompt = this.parseSlashCommandInput(payload.text) !== null;
 
@@ -1365,6 +1380,7 @@ export class PiRpcAgentSession implements AgentSession {
         this.pendingSettledMessages = null;
         this.activeAssistantMessageId = null;
         this.pendingSteerSubmissions.length = 0;
+        this.pendingForegroundSubmission = null;
         this.clearNoTurnBuffers();
         if (isPiRequestAbortError(error)) {
           this.emit({
@@ -1403,9 +1419,29 @@ export class PiRpcAgentSession implements AgentSession {
       return { status: "unavailable" };
     }
 
+    const imageCount = payload.images?.length ?? 0;
+    const isImageOnly = payload.text.trim().length === 0 && imageCount > 0;
+    let label = payload.text;
+    if (isImageOnly) {
+      if (imageCount === 1) {
+        label = "1 image";
+      } else {
+        label = `${imageCount} images`;
+      }
+    }
+    const submission: PiPendingSteerSubmission = {
+      text: payload.text,
+      label,
+      clientMessageId: options.clientMessageId ?? null,
+      consumed: false,
+    };
+    // Pi can consume the message before acknowledging the steer RPC.
+    this.pendingSteerSubmissions.push(submission);
     try {
       await this.runtimeSession.steer(payload.text, payload.images);
     } catch (error) {
+      const index = this.pendingSteerSubmissions.indexOf(submission);
+      if (index >= 0) this.pendingSteerSubmissions.splice(index, 1);
       if (isPiDefinitiveSteerRejection(error)) {
         return { status: "unavailable" };
       }
@@ -1416,10 +1452,6 @@ export class PiRpcAgentSession implements AgentSession {
     if (this.closed || this.activeTurnId !== options.expectedTurnId) {
       return { status: "unavailable" };
     }
-    this.pendingSteerSubmissions.push({
-      text: payload.text,
-      clientMessageId: options.clientMessageId ?? null,
-    });
     if (options.clearPendingPermissions) {
       await this.clearPendingPermissionsForSteer();
     }
@@ -1438,6 +1470,11 @@ export class PiRpcAgentSession implements AgentSession {
   }
 
   private takePendingSteerSubmission(text: string): PiPendingSteerSubmission | undefined {
+    // The original prompt is also echoed, possibly after a matching steer is queued.
+    if (this.pendingForegroundSubmission?.text === text) {
+      this.pendingForegroundSubmission = null;
+      return undefined;
+    }
     const index = this.pendingSteerSubmissions.findIndex((submission) => submission.text === text);
     if (index < 0) {
       return undefined;
@@ -1534,13 +1571,51 @@ export class PiRpcAgentSession implements AgentSession {
     };
   }
 
+  private reportClearedSteers(
+    cleared: PiClearedQueue | undefined,
+    turnId: string | undefined,
+  ): void {
+    if (!cleared) return;
+    // Pi returns what it actually removed, after any input-extension rewrites.
+    for (const text of cleared.steering) {
+      const index = this.pendingSteerSubmissions.findIndex(
+        (submission) => !submission.consumed && submission.text === text,
+      );
+      const submission = index >= 0 ? this.pendingSteerSubmissions.splice(index, 1)[0] : undefined;
+      // Pi retains consumed image-only entries in its text queue. An empty native
+      // entry proves removal only when we still have an unconsumed submission.
+      if (!text && !submission) continue;
+      const label = submission?.label ?? (text || "queued message");
+      this.emit({
+        type: "timeline",
+        provider: this.provider,
+        turnId,
+        item: {
+          type: "notification",
+          level: "warning",
+          message: `Not delivered (stopped): ${label}`,
+        },
+      });
+    }
+  }
+
   async interrupt(): Promise<void> {
     const turnId = this.activeTurnId ?? undefined;
     const interruption: typeof this.interruptingTurn =
       this.activeTurnId || this.activeTurnStarted ? { turnId, error: null } : null;
     this.interruptingTurn = interruption;
     try {
-      await this.abortRuntimeWork();
+      try {
+        const cleared = await this.runtimeSession.clearQueue();
+        this.reportClearedSteers(cleared, turnId);
+      } catch (error) {
+        // COMPAT(piClearQueueFallback): added in v0.5.0, remove after 2027-03-01 once
+        // the pi floor supports clear_queue (added in pi 0.84.4).
+        if (!isPiMissingClearQueueRpc(error)) {
+          throw error;
+        }
+      }
+      await this.runtimeSession.abort();
     } catch (error) {
       const terminalError = this.interruptingTurn === interruption ? interruption?.error : null;
       if (this.interruptingTurn === interruption) {
@@ -1555,6 +1630,7 @@ export class PiRpcAgentSession implements AgentSession {
         this.pendingSettledMessages = null;
         this.activeAssistantMessageId = null;
         this.pendingSteerSubmissions.length = 0;
+        this.pendingForegroundSubmission = null;
         this.clearNoTurnBuffers();
         this.emit({
           type: "turn_failed",
@@ -1579,6 +1655,7 @@ export class PiRpcAgentSession implements AgentSession {
       this.pendingSettledMessages = null;
       this.activeAssistantMessageId = null;
       this.pendingSteerSubmissions.length = 0;
+      this.pendingForegroundSubmission = null;
       this.clearNoTurnBuffers();
       this.emit({
         type: "turn_canceled",
@@ -1592,19 +1669,6 @@ export class PiRpcAgentSession implements AgentSession {
     }
   }
 
-  private async abortRuntimeWork(): Promise<void> {
-    try {
-      await this.runtimeSession.clearQueue();
-    } catch (error) {
-      // COMPAT(piClearQueueFallback): added in v0.5.0, remove after 2027-03-01 once
-      // the pi floor supports clear_queue (added in pi 0.84.4).
-      if (!isPiMissingClearQueueRpc(error)) {
-        throw error;
-      }
-    }
-    await this.runtimeSession.abort();
-  }
-
   async revertConversation(input: { messageId: string }): Promise<void> {
     if (this.activeTurnId) {
       throw new Error("Cannot rewind the Pi conversation while a turn is active");
@@ -1659,13 +1723,6 @@ export class PiRpcAgentSession implements AgentSession {
     this.extensionHost.close();
     this.usagePoller.close();
     try {
-      if (this.activeTurnId || this.activeTurnStarted) {
-        // Pi persists terminal tool results before acknowledging abort. Keep its
-        // event stream alive until then so restart history cannot strand the tool.
-        await this.abortRuntimeWork().catch((err) => {
-          this.logger.warn({ err }, "Failed to abort Pi work before closing its runtime");
-        });
-      }
       await this.runtimeSession.close();
     } finally {
       await Promise.all([this.pendingOutput, ...this.pendingExtensionHydrations]);
@@ -2214,6 +2271,7 @@ export class PiRpcAgentSession implements AgentSession {
     this.activeTurnStartedEmitted = false;
     this.pendingSettledMessages = null;
     this.pendingSteerSubmissions.length = 0;
+    this.pendingForegroundSubmission = null;
     this.clearNoTurnBuffers();
     this.emit({
       type: "turn_failed",
@@ -2492,6 +2550,21 @@ export class PiRpcAgentSession implements AgentSession {
     event: Extract<PiAgentSessionEvent, { type: "message_end" }>,
     turnId: string | undefined,
   ): void {
+    if (event.message.role === "user") {
+      // Entry IDs arrive through the extension later; consumption is already known.
+      const text = getUserMessageText(event.message.content);
+      const foreground = this.pendingForegroundSubmission;
+      // prompt() runs asynchronously, so its first user event may follow steer().
+      if (foreground && !foreground.consumed && foreground.text === text) {
+        foreground.consumed = true;
+        return;
+      }
+      const submission = this.pendingSteerSubmissions.find(
+        (pending) => !pending.consumed && pending.text === text,
+      );
+      if (submission) submission.consumed = true;
+      return;
+    }
     if (event.message.role === "assistant") {
       this.activeAssistantMessageId = null;
       return;
@@ -2574,6 +2647,7 @@ export class PiRpcAgentSession implements AgentSession {
     this.activeTurnStartedEmitted = false;
     this.pendingSettledMessages = null;
     this.pendingSteerSubmissions.length = 0;
+    this.pendingForegroundSubmission = null;
     this.clearNoTurnBuffers();
     if (typeof errorMessage === "string" && errorMessage.length > 0) {
       this.usagePoller.stopTurn();

packages/server/src/server/agent/providers/pi/cli-runtime.test.ts

@@ -419,11 +419,12 @@ describe("PiCliRuntime", () => {
 
   test("sends the clear_queue RPC frame", async () => {
     const child = createPiChild();
-    replyToCommands(child, () => ({}));
+    const removed = { steering: ["rewritten message"], followUp: [] };
+    replyToCommands(child, () => removed);
     const session = await createRuntime(child).startSession({ cwd: "/workspace/project" });
 
     const clearQueueCommand = capturePendingCommand(child, "clear_queue");
-    await session.clearQueue();
+    await expect(session.clearQueue()).resolves.toEqual(removed);
 
     expect(await clearQueueCommand).toMatchObject({ type: "clear_queue" });
   });

packages/server/src/server/agent/providers/pi/cli-runtime.ts

@@ -12,10 +12,12 @@ import {
 } from "../jsonl-rpc-process.js";
 import {
   buildPiLaunch,
+  PiClearedQueueSchema,
   type PiRuntime,
   type PiRuntimeLaunch,
   type PiRuntimeSession,
   type PiStartSessionInput,
+  type PiClearedQueue,
 } from "./runtime.js";
 import type {
   PiModel,
@@ -149,8 +151,10 @@ class PiCliRuntimeSession implements PiRuntimeSession {
     });
   }
 
-  async clearQueue(): Promise<void> {
-    await this.requestStopWork({ type: "clear_queue" });
+  async clearQueue(): Promise<PiClearedQueue | undefined> {
+    const result = await this.options.process.requestStopWork({ type: "clear_queue" });
+    if (result === undefined) return undefined;
+    return PiClearedQueueSchema.parse(result);
   }
 
   async compact(customInstructions?: string): Promise<void> {
@@ -268,8 +272,8 @@ class PiCliRuntimeSession implements PiRuntimeSession {
     return this.options.process.request(command, timeoutMs);
   }
 
-  private requestStopWork(command: PiRpcCommand): Promise<void> {
-    return this.options.process.requestStopWork(command);
+  private async requestStopWork(command: PiRpcCommand): Promise<void> {
+    await this.options.process.requestStopWork(command);
   }
 
   private async waitForCompletion(command: PiRpcCommand): Promise<void> {

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

@@ -1,4 +1,5 @@
 import type { PiTranscript } from "./transcript.js";
+import { z } from "zod";
 import type {
   PiModel,
   PiPromptAck,
@@ -39,6 +40,12 @@ export interface PiStartSessionInput {
   extraArgs?: string[];
 }
 
+export const PiClearedQueueSchema = z.object({
+  steering: z.array(z.string()),
+  followUp: z.array(z.string()),
+});
+export type PiClearedQueue = z.infer<typeof PiClearedQueueSchema>;
+
 export interface PiRuntimeSession {
   readonly environment: Record<string, string>;
   onEvent(callback: (event: PiRuntimeEvent) => void): () => void;
@@ -50,7 +57,7 @@ export interface PiRuntimeSession {
     message: string,
     images?: Array<{ type: "image"; data: string; mimeType: string }>,
   ): Promise<void>;
-  clearQueue(): Promise<void>;
+  clearQueue(): Promise<PiClearedQueue | undefined>;
   compact(customInstructions?: string): Promise<void>;
   setAutoCompaction(enabled: boolean): Promise<void>;
   abort(): Promise<void>;

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

@@ -5,6 +5,7 @@ import type {
   PiRuntimeLaunch,
   PiRuntimeSession,
   PiStartSessionInput,
+  PiClearedQueue,
 } from "../runtime.js";
 import type {
   PiAgentMessage,
@@ -16,6 +17,7 @@ import type {
   PiSessionStats,
 } from "../rpc-types.js";
 import { buildPiLaunch } from "../runtime.js";
+import { getUserMessageText } from "../history-mapper.js";
 
 type FakePiSubagentSubscriptionLevel = "off" | "progress" | "events";
 type FakePiSubagentStatus = "pending" | "running" | "completed" | "failed" | "aborted";
@@ -135,6 +137,7 @@ export class FakePiSession implements PiRuntimeSession {
   readonly prompts: Array<{ message: string; imageCount: number }> = [];
   readonly steerCalls: Array<{ message: string; imageCount: number }> = [];
   steerError: Error | null = null;
+  steerTextTransform: (text: string) => string = (text) => text;
   readonly controlRequests: string[] = [];
   clearQueueError: Error | null = null;
   clearQueueCompletion: Promise<void> | null = null;
@@ -153,7 +156,6 @@ export class FakePiSession implements PiRuntimeSession {
   // Native user entries paired with the complete active-branch history.
   historyUserEntries: FakePiUserEntry[] = [];
   abortRequested = false;
-  abortCompletion: Promise<void> | null = null;
   closeCalls = 0;
   readonly canceledExtensionUiRequests: string[] = [];
   readonly extensionUiResponses: Array<{
@@ -183,6 +185,8 @@ export class FakePiSession implements PiRuntimeSession {
   state: PiSessionState;
 
   private readonly subscribers = new Set<(event: PiRuntimeEvent) => void>();
+  private readonly steeringMessages: string[] = [];
+  private foregroundMessage: string | null = null;
   private readonly subagentMessageResults = new Map<string, FakePiSubagentMessagesResult[]>();
   private nextHeldPrompt: { promise: Promise<void>; reject: (error: Error) => void } | null = null;
   private activeHeldPrompt: { promise: Promise<void>; reject: (error: Error) => void } | null =
@@ -218,6 +222,7 @@ export class FakePiSession implements PiRuntimeSession {
     images?: Array<{ type: "image"; data: string; mimeType: string }>,
   ): Promise<PiPromptAck> {
     this.prompts.push({ message, imageCount: images?.length ?? 0 });
+    if (!message.startsWith("/")) this.foregroundMessage = message;
     const heldPrompt = this.nextHeldPrompt;
     if (heldPrompt) {
       this.nextHeldPrompt = null;
@@ -263,14 +268,16 @@ export class FakePiSession implements PiRuntimeSession {
     if (this.steerError) {
       throw this.steerError;
     }
+    this.steeringMessages.push(this.steerTextTransform(message));
   }
 
-  async clearQueue(): Promise<void> {
+  async clearQueue(): Promise<PiClearedQueue> {
     this.controlRequests.push("clear_queue");
     await this.clearQueueCompletion;
     if (this.clearQueueError) {
       throw this.clearQueueError;
     }
+    return { steering: this.steeringMessages.splice(0), followUp: [] };
   }
 
   async compact(customInstructions?: string): Promise<void> {
@@ -296,7 +303,6 @@ export class FakePiSession implements PiRuntimeSession {
   async abort(): Promise<void> {
     this.controlRequests.push("abort");
     this.abortRequested = true;
-    await this.abortCompletion;
   }
 
   async getState(): Promise<PiSessionState> {
@@ -429,10 +435,19 @@ export class FakePiSession implements PiRuntimeSession {
 
   async close(): Promise<void> {
     this.closeCalls += 1;
-    this.subscribers.clear();
   }
 
   emit(event: PiRuntimeEvent): void {
+    const message = event.type === "message_end" ? (event.message as PiAgentMessage) : null;
+    if (message?.role === "user") {
+      const text = getUserMessageText(message.content);
+      if (this.foregroundMessage === text) {
+        this.foregroundMessage = null;
+      } else if (text) {
+        const index = this.steeringMessages.indexOf(text);
+        if (index >= 0) this.steeringMessages.splice(index, 1);
+      }
+    }
     for (const subscriber of this.subscribers) {
       subscriber(event);
     }

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

@@ -1,6 +1,8 @@
 import { createWorkspaceRuntimeStore } from "./workspace-runtime-store/index.js";
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { existsSync } from "node:fs";
+import { performance } from "node:perf_hooks";
+import { setTimeout as delay } from "node:timers/promises";
 import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
 import os from "node:os";
 import path, { join } from "node:path";
@@ -421,20 +423,20 @@ function linkedWorktreeFacts(gitCommonDir: string) {
   });
 }
 
-const SETTLE_ATTEMPTS = 1_000;
-
 async function settle(assertion: () => void): Promise<void> {
-  let lastFailure: unknown = null;
-  for (let attempt = 0; attempt < SETTLE_ATTEMPTS; attempt += 1) {
+  // Observation setup resolves real filesystem paths even while the polling clock is frozen.
+  // Give that I/O wall-clock time without advancing the service's polling deadlines.
+  const deadline = performance.now() + 5_000;
+  while (true) {
     try {
       assertion();
       return;
     } catch (failure) {
-      lastFailure = failure;
+      if (performance.now() >= deadline) throw failure;
     }
+    await delay(1);
     await vi.advanceTimersByTimeAsync(0);
   }
-  throw lastFailure;
 }
 
 async function waitForRefreshesToSettle(service: WorkspaceGitServiceImpl): Promise<void> {
@@ -1743,48 +1745,65 @@ describe("WorkspaceGitServiceImpl with workspace folders that no longer exist",
     await service.dispose();
   });
 
-  test("a deleted worktree leaves its shared repository polling and fetch to the surviving worktree", async () => {
-    const gitCommonDir = join(await createTemporaryFolder(), ".git");
-    const deletedWorktree = await createTemporaryFolder();
-    const survivingWorktree = await createTemporaryFolder();
-    const git = createFolderBackedGit({ facts: linkedWorktreeFacts(gitCommonDir) });
-    let survivingWorktreeEdits = 0;
-    git.deps.getCheckoutShortstat.mockImplementation(async (cwd: string) => ({
-      additions: cwd === survivingWorktree ? ++survivingWorktreeEdits : 1,
-      deletions: 0,
-    }));
-    const service = createService(git.deps);
-    const deletedSubscription = service.registerWorkspace({ cwd: deletedWorktree }, vi.fn());
-    await waitForObservationToSettle(service, 1);
-    const survivingSubscription = service.registerWorkspace({ cwd: survivingWorktree }, vi.fn());
-    await waitForObservationToSettle(service, 2);
-
-    await rm(deletedWorktree, { recursive: true, force: true });
-    await vi.advanceTimersByTimeAsync(60_000);
-    await waitForRefreshesToSettle(service);
-    const deletedWorktreeCommands = git.gitCommandsIn(deletedWorktree);
-
-    for (let tick = 0; tick < 3; tick += 1) {
-      const survivingWorktreePolls = git.pollsOf(survivingWorktree);
-      await vi.advanceTimersByTimeAsync(5_000);
-      await settle(() => {
-        expect(git.pollsOf(survivingWorktree)).toBe(survivingWorktreePolls + 1);
+  test.each([0, 250])(
+    "a deleted worktree leaves its shared repository polling and fetch to the surviving worktree (watch delay: %ims)",
+    async (watchDelayMs) => {
+      const gitCommonDir = join(await createTemporaryFolder(), ".git");
+      const deletedWorktree = await createTemporaryFolder();
+      const survivingWorktree = await createTemporaryFolder();
+      const watchReady = createDeferred<void>();
+      const git = createFolderBackedGit({
+        facts: linkedWorktreeFacts(gitCommonDir),
+        heldWatch: { directory: survivingWorktree, release: watchReady.promise },
       });
-    }
-    expect(git.gitCommandsIn(deletedWorktree)).toBe(deletedWorktreeCommands);
+      let survivingWorktreeEdits = 0;
+      git.deps.getCheckoutShortstat.mockImplementation(async (cwd: string) => ({
+        additions: cwd === survivingWorktree ? ++survivingWorktreeEdits : 1,
+        deletions: 0,
+      }));
+      const service = createService(git.deps);
+      try {
+        const deletedSubscription = service.registerWorkspace({ cwd: deletedWorktree }, vi.fn());
+        await waitForObservationToSettle(service, 1);
+        const survivingSubscription = service.registerWorkspace(
+          { cwd: survivingWorktree },
+          vi.fn(),
+        );
+        const pollingTime = Date.now();
+        void delay(watchDelayMs).then(() => watchReady.resolve());
+        await waitForObservationToSettle(service, 2);
+        expect(Date.now()).toBe(pollingTime);
 
-    const fetchesBefore = git.deps.runGitFetch.mock.calls.length;
-    await vi.advanceTimersByTimeAsync(2 * BACKGROUND_GIT_FETCH_INTERVAL_MS);
-    expect(git.deps.runGitFetch.mock.calls.slice(fetchesBefore).map(([cwd]) => cwd)).toEqual([
-      survivingWorktree,
-      survivingWorktree,
-    ]);
-    expect(git.gitCommandsIn(deletedWorktree)).toBe(deletedWorktreeCommands);
+        await rm(deletedWorktree, { recursive: true, force: true });
+        await vi.advanceTimersByTimeAsync(60_000);
+        await waitForRefreshesToSettle(service);
+        const deletedWorktreeCommands = git.gitCommandsIn(deletedWorktree);
 
-    deletedSubscription.unsubscribe();
-    survivingSubscription.unsubscribe();
-    await service.dispose();
-  });
+        for (let tick = 0; tick < 3; tick += 1) {
+          const survivingWorktreePolls = git.pollsOf(survivingWorktree);
+          await vi.advanceTimersByTimeAsync(5_000);
+          await settle(() => {
+            expect(git.pollsOf(survivingWorktree)).toBe(survivingWorktreePolls + 1);
+          });
+        }
+        expect(git.gitCommandsIn(deletedWorktree)).toBe(deletedWorktreeCommands);
+
+        const fetchesBefore = git.deps.runGitFetch.mock.calls.length;
+        await vi.advanceTimersByTimeAsync(2 * BACKGROUND_GIT_FETCH_INTERVAL_MS);
+        expect(git.deps.runGitFetch.mock.calls.slice(fetchesBefore).map(([cwd]) => cwd)).toEqual([
+          survivingWorktree,
+          survivingWorktree,
+        ]);
+        expect(git.gitCommandsIn(deletedWorktree)).toBe(deletedWorktreeCommands);
+
+        deletedSubscription.unsubscribe();
+        survivingSubscription.unsubscribe();
+      } finally {
+        watchReady.resolve();
+        await service.dispose();
+      }
+    },
+  );
 
   test("deleting every worktree of a repository stops its polling and background fetch", async () => {
     const gitCommonDir = join(await createTemporaryFolder(), ".git");