fix(ios): hold a canceled request's locks until its runner command ends

packages/platform-apple/src/runner/__tests__/fake-runner-server.ts

@@ -17,7 +17,9 @@ export type FakeRunnerResponse =
   /** Hangs up on this request and every later one for the command: the entry is never consumed. */
   | { kind: 'hangUpAlways' }
   /** Hangs up and stops listening, as a runner process that died mid-command. */
-  | { kind: 'exit' };
+  | { kind: 'exit' }
+  /** Never answers, as a runner still executing the command when its caller gives up. */
+  | { kind: 'hold' };
 
 export type FakeRunnerRequest = {
   command: string;
@@ -79,9 +81,10 @@ export async function startFakeRunnerServer(
     requests,
     close: () =>
       stopped ??
-      new Promise<void>((resolve, reject) =>
-        server.close((error) => (error ? reject(error) : resolve())),
-      ),
+      new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+        server.closeAllConnections();
+      }),
   };
 }
 
@@ -101,6 +104,7 @@ function writeFakeRunnerResponse(
     res.end(JSON.stringify({ ok: false, error: { message: 'fake runner script exhausted' } }));
     return;
   }
+  if (next.kind === 'hold') return;
   if (next.kind === 'hangUp' || next.kind === 'hangUpAlways') {
     res.destroy();
     return;

packages/platform-apple/src/runner/__tests__/runner-exchange.test.ts

@@ -1,10 +1,25 @@
 import assert from 'node:assert/strict';
 import type { ExecResult } from '@agent-device/host-kit/command';
-import { afterEach, test } from 'vitest';
+import { createRequestCanceledError, isRequestCanceledError } from '@agent-device/kernel/errors';
+import { withKeyedLock } from '@agent-device/kernel/keyed-lock';
+import { afterEach, test, vi } from 'vitest';
+
+const { mockSendRunnerCommandOnce } = vi.hoisted(() => ({ mockSendRunnerCommandOnce: vi.fn() }));
+
+vi.mock('../runner-transport.ts', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('../runner-transport.ts')>();
+  mockSendRunnerCommandOnce.mockImplementation(actual.sendRunnerCommandOnce);
+  return { ...actual, sendRunnerCommandOnce: mockSendRunnerCommandOnce };
+});
+
 import { executeRunnerExchange } from '../runner-exchange.ts';
 import { RunnerCommandAccounting, type RunnerSession } from '../runner-session-types.ts';
 import { IOS_SIMULATOR } from './device-fixtures.ts';
-import { startFakeRunnerServer, type FakeRunnerServer } from './fake-runner-server.ts';
+import {
+  startFakeRunnerServer,
+  type FakeRunnerResponse,
+  type FakeRunnerServer,
+} from './fake-runner-server.ts';
 
 let server: FakeRunnerServer | undefined;
 
@@ -184,3 +199,233 @@ test('the exchange awaits owner invalidation before throwing a fatal runner erro
   }
   await assert.rejects(exchange);
 });
+
+const LONG_PRESS = {
+  command: 'longPress',
+  x: 10,
+  y: 10,
+  durationMs: 5_000,
+  appBundleId: 'com.example.app',
+} as const;
+
+function journalState(lifecycleState: string): FakeRunnerResponse {
+  return { kind: 'ok', data: { lifecycleState } };
+}
+
+async function abortOnceReceived(
+  runner: FakeRunnerServer,
+  command: string,
+  controller: AbortController,
+  reason: unknown = createRequestCanceledError(),
+): Promise<void> {
+  const giveUpAtMs = Date.now() + 2_000;
+  while (!runner.requests.some((request) => request.command === command)) {
+    if (Date.now() >= giveUpAtMs) {
+      throw new Error(`The fake runner never received "${command}"; nothing to abort.`);
+    }
+    await new Promise<void>((resolve) => setTimeout(resolve, 1));
+  }
+  controller.abort(reason);
+}
+
+function statusPolls(runner: FakeRunnerServer): unknown[] {
+  return runner.requests
+    .filter((request) => request.command === 'status')
+    .map((request) => request.body.statusCommandId);
+}
+
+function sentCommandId(runner: FakeRunnerServer, command: string): unknown {
+  return runner.requests.find((request) => request.command === command)?.body.commandId;
+}
+
+test('a canceled exchange whose command reached the runner settles only once its journal entry is terminal', async () => {
+  server = await startFakeRunnerServer({
+    longPress: [{ kind: 'hold' }],
+    status: [journalState('started'), journalState('completed')],
+  });
+  const runner = server;
+  const session = sessionFor(runner.port);
+  const controller = new AbortController();
+  let requestsWhenSettled = -1;
+
+  const exchange = executeRunnerExchange(
+    IOS_SIMULATOR,
+    session,
+    LONG_PRESS,
+    undefined,
+    10_000,
+    async () => {},
+    controller.signal,
+  ).finally(() => {
+    requestsWhenSettled = runner.requests.length;
+  });
+  await abortOnceReceived(runner, 'longPress', controller);
+
+  await assert.rejects(exchange, isRequestCanceledError);
+  const commandId = sentCommandId(runner, 'longPress');
+  assert.deepEqual(statusPolls(runner), [commandId, commandId]);
+  assert.equal(requestsWhenSettled, runner.requests.length);
+  assert.equal(session.commandCharges.hasOutstandingCharges, false);
+});
+
+test('a request on the same session lock starts only after the canceled command is terminal', async () => {
+  server = await startFakeRunnerServer({
+    longPress: [{ kind: 'hold' }],
+    status: [journalState('accepted'), journalState('started'), journalState('failed')],
+  });
+  const runner = server;
+  const session = sessionFor(runner.port);
+  const locks = new Map<string, Promise<unknown>>();
+  const controller = new AbortController();
+
+  const canceled = withKeyedLock(locks, 'session:default', async () => {
+    await executeRunnerExchange(
+      IOS_SIMULATOR,
+      session,
+      LONG_PRESS,
+      undefined,
+      10_000,
+      async () => {},
+      controller.signal,
+    );
+  });
+  await abortOnceReceived(runner, 'longPress', controller);
+  const nextRequestSawPolls = withKeyedLock(locks, 'session:default', async () =>
+    statusPolls(runner),
+  );
+
+  await assert.rejects(canceled, isRequestCanceledError);
+  assert.equal((await nextRequestSawPolls).length, 3);
+  assert.equal(session.commandCharges.hasOutstandingCharges, false);
+});
+
+test('a request canceled before its command is sent releases without reading the journal', async () => {
+  server = await startFakeRunnerServer({});
+  const session = sessionFor(server.port);
+  session.lastHealthyMutation = { atMs: Date.now(), appBundleId: LONG_PRESS.appBundleId };
+  const controller = new AbortController();
+  controller.abort(createRequestCanceledError());
+
+  await assert.rejects(
+    executeRunnerExchange(
+      IOS_SIMULATOR,
+      session,
+      LONG_PRESS,
+      undefined,
+      10_000,
+      async () => {},
+      controller.signal,
+    ),
+    isRequestCanceledError,
+  );
+
+  assert.deepEqual(server.requests, []);
+  assert.equal(session.commandCharges.hasOutstandingCharges, false);
+});
+
+test('a cancellation the transport proves unsent withdraws the charge without reading the journal', async () => {
+  server = await startFakeRunnerServer({ status: [journalState('started')] });
+  const session = sessionFor(server.port);
+  session.lastHealthyMutation = { atMs: Date.now(), appBundleId: LONG_PRESS.appBundleId };
+  const controller = new AbortController();
+  mockSendRunnerCommandOnce.mockImplementationOnce(async () => {
+    controller.abort(createRequestCanceledError());
+    throw createRequestCanceledError({ dispatched: 'no' });
+  });
+
+  await assert.rejects(
+    executeRunnerExchange(
+      IOS_SIMULATOR,
+      session,
+      LONG_PRESS,
+      undefined,
+      10_000,
+      async () => {},
+      controller.signal,
+    ),
+    isRequestCanceledError,
+  );
+
+  assert.deepEqual(server.requests, []);
+  assert.equal(session.commandCharges.hasOutstandingCharges, false);
+});
+
+test('a canceled command the journal never accepted releases after one read and stays abandoned', async () => {
+  server = await startFakeRunnerServer({
+    longPress: [{ kind: 'hold' }],
+    status: [journalState('notAccepted')],
+  });
+  const session = sessionFor(server.port);
+  const controller = new AbortController();
+
+  const exchange = executeRunnerExchange(
+    IOS_SIMULATOR,
+    session,
+    LONG_PRESS,
+    undefined,
+    10_000,
+    async () => {},
+    controller.signal,
+  );
+  await abortOnceReceived(server, 'longPress', controller);
+
+  await assert.rejects(exchange, isRequestCanceledError);
+  assert.equal(statusPolls(server).length, 1);
+  assert.equal(session.commandCharges.hasAbandonedCharges, true);
+});
+
+test('a drain that outlives the command deadline settles and leaves the charge abandoned', async () => {
+  server = await startFakeRunnerServer({
+    longPress: [{ kind: 'hold' }],
+    status: Array.from({ length: 50 }, () => journalState('started')),
+  });
+  const session = sessionFor(server.port);
+  const controller = new AbortController();
+  const timeoutMs = 300;
+  const startedAtMs = Date.now();
+
+  const exchange = executeRunnerExchange(
+    IOS_SIMULATOR,
+    session,
+    LONG_PRESS,
+    undefined,
+    timeoutMs,
+    async () => {},
+    controller.signal,
+  );
+  await abortOnceReceived(server, 'longPress', controller);
+
+  await assert.rejects(exchange, isRequestCanceledError);
+  assert.ok(statusPolls(server).length >= 1);
+  assert.ok(Date.now() - startedAtMs < timeoutMs + 1_000);
+  assert.equal(session.commandCharges.hasAbandonedCharges, true);
+});
+
+test("a caller's own deadline abort does not wait on the runner journal", async () => {
+  server = await startFakeRunnerServer({
+    longPress: [{ kind: 'hold' }],
+    status: [journalState('started'), journalState('completed')],
+  });
+  const session = sessionFor(server.port);
+  const controller = new AbortController();
+
+  const exchange = executeRunnerExchange(
+    IOS_SIMULATOR,
+    session,
+    LONG_PRESS,
+    undefined,
+    10_000,
+    async () => {},
+    controller.signal,
+  );
+  await abortOnceReceived(
+    server,
+    'longPress',
+    controller,
+    new DOMException('Wait deadline exceeded', 'TimeoutError'),
+  );
+
+  await assert.rejects(exchange);
+  assert.deepEqual(statusPolls(server), []);
+  assert.equal(session.commandCharges.hasAbandonedCharges, true);
+});

packages/platform-apple/src/runner/__tests__/runner-session-types.test.ts

@@ -237,6 +237,19 @@ describe('RunnerCommandAccounting abandonment', () => {
   });
 });
 
+describe('RunnerCommandAccounting unsent sends', () => {
+  test('withdraws only the charge of the send that wrote nothing', () => {
+    const charges = new RunnerCommandAccounting();
+    charges.charge('cmd-a');
+    charges.markAbandoned('cmd-a');
+    charges.charge('cmd-b');
+    charges.withdrawUnsent('cmd-b');
+
+    expect(charges.outstandingChargeCount).toBe(1);
+    expect(charges.hasAbandonedCharges).toBe(true);
+  });
+});
+
 describe('RunnerCommandAccounting terminal evidence', () => {
   test('discharges the abandoned charge it names', () => {
     const charges = new RunnerCommandAccounting();

packages/platform-apple/src/runner/__tests__/runner-transport.test.ts

@@ -2,7 +2,11 @@ import { afterEach, beforeEach, test, vi } from 'vitest';
 import assert from 'node:assert/strict';
 import fs from 'node:fs';
 import path from 'node:path';
-import { AppError } from '@agent-device/kernel/errors';
+import {
+  AppError,
+  createRequestCanceledError,
+  isRequestCanceledError,
+} from '@agent-device/kernel/errors';
 import {
   iosDevice,
   iosSimulator,
@@ -221,3 +225,31 @@ test('falls back to the tunnel when usbmux loses the device mid-connect', async
   assert.equal(response.status, 200);
   assert.equal(vi.mocked(fetch).mock.calls[0]?.[0], 'http://[fd00::123]:8100/command');
 });
+
+test('a cancellation during route resolution discloses that the command was never sent', async () => {
+  mockUsbmuxPostCommand.mockRejectedValue(usbmuxDeviceUnattachedError());
+  stubSuccessfulFetch();
+  const controller = new AbortController();
+  const lookupTunnelIp = makeTunnelIpLookup('fd00::123');
+  mockRunCmd.mockImplementation(async (cmd: string, args: string[]) => {
+    controller.abort(createRequestCanceledError());
+    return await lookupTunnelIp(cmd, args);
+  });
+
+  await assert.rejects(
+    () =>
+      sendRunnerCommandOnce(
+        iosDevice,
+        8100,
+        { command: 'tap', x: 1, y: 1 },
+        5_000,
+        controller.signal,
+      ),
+    (error: unknown) => {
+      assert.ok(isRequestCanceledError(error));
+      assert.equal((error as AppError).details?.dispatched, 'no');
+      return true;
+    },
+  );
+  assert.equal(vi.mocked(fetch).mock.calls.length, 0);
+});