fix(pi): settle foreground tools before runtime shutdown

Wait for native work to abort while the event stream is still open so Pi persists terminal tool results before restart history is loaded.

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 input
+## Inverted timeline selection
 
 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,10 +183,6 @@ 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.

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

@@ -4,13 +4,11 @@ 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
@@ -26,40 +24,6 @@ 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())
@@ -106,7 +70,7 @@ class ScrollSelectionTest {
       val appContext = instrumentation.targetContext
       DisplayMetricsHolder.initDisplayMetricsIfNotInitialized(appContext)
       SoLoader.init(appContext, OpenSourceMergedSoMapping)
-      val reactContext = ScrollTestContext(appContext)
+      val reactContext = BridgeReactContext(appContext)
       val context = ThemedReactContext(reactContext, appContext)
       val manager = PaseoScrollPackage().createViewManager(reactContext, ReactScrollViewManager.REACT_CLASS) as ReactScrollViewManager
       val scroll = manager.createViewInstance(context)
@@ -134,26 +98,4 @@ 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

@@ -1,19 +0,0 @@
-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,8 +2,6 @@ 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
@@ -32,47 +30,13 @@ 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 =
-    if (isVerticallyInverted()) 0 else super.computeScrollDeltaToGetChildRectOnScreen(rect)
+  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)
+  }
 }

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

@@ -58,9 +58,6 @@ 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");
 
@@ -673,33 +670,6 @@ 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/opencode/v2/session.ts

@@ -242,10 +242,6 @@ 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,6 +208,52 @@ 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);
@@ -2183,48 +2229,6 @@ describe("PiRpcAgentSession", () => {
     ]);
   });
 
-  test.each([false, true])(
-    "closing Pi releases an active turn without publishing a terminal outcome (autonomous=%s)",
-    async (autonomous) => {
-      const { pi, session, events } = await createSession();
-      if (autonomous) {
-        pi.latestSession().emit({ type: "agent_start" });
-        pi.latestSession().emit({ type: "turn_start" });
-      } else {
-        await session.startTurn("keep working");
-      }
-      const before = events.turnLifecycleEvents();
-
-      await session.close();
-
-      expect(pi.latestSession().closeCalls).toBe(1);
-      expect(events.turnLifecycleEvents()).toEqual(before);
-    },
-  );
-
-  test("closing Pi before prompt acknowledgement preserves the unfinished turn", async () => {
-    const { pi, session, events } = await createSession();
-    pi.latestSession().holdNextPrompt();
-    await session.startTurn("keep working");
-    expect(pi.latestSession().prompts).toHaveLength(1);
-    const before = events.turnLifecycleEvents();
-
-    await session.close();
-
-    expect(events.turnLifecycleEvents()).toEqual(before);
-  });
-
-  test("a rejected prompt still fails an open Pi session", async () => {
-    const { pi, session, events } = await createSession();
-    pi.latestSession().holdNextPrompt();
-    await session.startTurn("keep working");
-
-    await pi.latestSession().failHeldPrompt(new Error("Prompt rejected"));
-
-    await expect(events.nextTurnFailure()).resolves.toMatchObject({ error: "Prompt rejected" });
-    await session.close();
-  });
-
   test("fails the active turn when the Pi process exits mid-turn", async () => {
     const { pi, session, events } = await createSession();
 

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

@@ -1540,16 +1540,7 @@ export class PiRpcAgentSession implements AgentSession {
       this.activeTurnId || this.activeTurnStarted ? { turnId, error: null } : null;
     this.interruptingTurn = interruption;
     try {
-      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();
+      await this.abortRuntimeWork();
     } catch (error) {
       const terminalError = this.interruptingTurn === interruption ? interruption?.error : null;
       if (this.interruptingTurn === interruption) {
@@ -1601,6 +1592,19 @@ 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");
@@ -1655,6 +1659,13 @@ 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]);
@@ -1729,17 +1740,6 @@ export class PiRpcAgentSession implements AgentSession {
   }
 
   private publish(event: AgentStreamEvent): void {
-    // Runtime release preserves unfinished work for restart recovery. Cleanup
-    // may still emit timeline updates, but must not settle or restart its turn.
-    if (
-      this.closed &&
-      (event.type === "turn_started" ||
-        event.type === "turn_completed" ||
-        event.type === "turn_canceled" ||
-        event.type === "turn_failed")
-    ) {
-      return;
-    }
     for (const subscriber of this.subscribers) subscriber(event);
   }
 

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

@@ -153,6 +153,7 @@ 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<{
@@ -295,6 +296,7 @@ export class FakePiSession implements PiRuntimeSession {
   async abort(): Promise<void> {
     this.controlRequests.push("abort");
     this.abortRequested = true;
+    await this.abortCompletion;
   }
 
   async getState(): Promise<PiSessionState> {
@@ -427,13 +429,7 @@ export class FakePiSession implements PiRuntimeSession {
 
   async close(): Promise<void> {
     this.closeCalls += 1;
-    if (this.activeHeldPrompt) {
-      await this.failHeldPrompt(new Error("Pi RPC session is closed"));
-    }
-    this.emit({
-      type: "process_exit",
-      error: "Pi RPC process exited with code 0 and signal null",
-    });
+    this.subscribers.clear();
   }
 
   emit(event: PiRuntimeEvent): void {