fix(capture-kit): make a completed durable record terminal

packages/capture-kit/src/app-log-live-handle.ts

@@ -8,6 +8,7 @@ import {
   type CleanupOutcome,
   type FinishOutcome,
   isConfirmedCleanup,
+  createRetryableCleanup,
 } from '@agent-device/contracts/durable-resource';
 
 type AppLogLiveHandleImplementation = Readonly<{
@@ -21,17 +22,10 @@ export function createAppLogLiveHandle(
   implementation: AppLogLiveHandleImplementation,
 ): AppLogLiveHandle {
   let finish: Promise<FinishOutcome<AppLogCompletion>> | undefined;
-  let cleanup: Promise<CleanupOutcome> | undefined;
-  let disposal: Promise<void> | undefined;
-  const forceCleanup = (): Promise<CleanupOutcome> => (cleanup ??= implementation.forceCleanup());
   return Object.freeze({
     inspect: () => implementation.inspect(),
     finish: () => (finish ??= implementation.finish()),
-    forceCleanup,
-    [Symbol.asyncDispose]: async () => {
-      disposal ??= forceCleanup().then(assertConfirmedCleanup);
-      await disposal;
-    },
+    ...createRetryableCleanup(() => implementation.forceCleanup(), assertConfirmedCleanup),
   });
 }
 

packages/capture-kit/src/app-log-pid-process.test.ts

@@ -8,6 +8,27 @@ import { deferred, processFixture } from './app-log-pid-process.fixtures.ts';
 import { createPidScopedAppLogProcess } from './app-log-pid-process.ts';
 
 describe('PID-scoped app-log process lifecycle', () => {
+  test('retries output disposal after an unconfirmed finish', async () => {
+    const output = outputFixture();
+    output.dispose.mockRejectedValueOnce(new Error('Close unavailable'));
+    const started =
+      processFixture(deferred<{ stdout: string; stderr: string; exitCode: number }>());
+    const handle = await createPidScopedAppLogProcess({
+      host: hostFixture(output.sink),
+      backend: 'android',
+      outputPath: '/tmp/app.log',
+      pidPath: '/tmp/pid',
+      processStart: async () => started.process,
+      setupSignal: new AbortController().signal,
+      resolvePid: async () => '123',
+      command: () => ({ kind: 'host', request: { executable: 'adb', args: ['logcat'] } }),
+      cleanupFailureMessage: 'cleanup failed',
+    });
+    await expect(handle.finish()).resolves.toMatchObject({ status: 'cleanup-pending' });
+    await handle[Symbol.asyncDispose]();
+    await handle[Symbol.asyncDispose]();
+    expect(output.dispose).toHaveBeenCalledTimes(2);
+  });
   test('cleans a later process that resolves after finish begins', async () => {
     const initialExit = deferred<{ stdout: string; stderr: string; exitCode: number }>();
     const laterExit = deferred<{ stdout: string; stderr: string; exitCode: number }>();

packages/capture-kit/src/app-log-pid-process.ts

@@ -55,14 +55,16 @@ export async function createPidScopedAppLogProcess(
       state = 'active';
     }
   } catch (error) {
-    await settleCleanupSteps([
-      async () => await active?.terminate(),
-      async () => {
-        await active?.wait;
-      },
-      async () => await active?.[Symbol.asyncDispose](),
-      async () => await output[Symbol.asyncDispose](),
-    ]);
+    await settleCleanupSteps(
+      new Set([
+        async () => await active?.terminate(),
+        async () => {
+          await active?.wait;
+        },
+        async () => await active?.[Symbol.asyncDispose](),
+        async () => await output[Symbol.asyncDispose](),
+      ]),
+    );
     throw error;
   }
   const monitor = monitorPidScopedProcess({
@@ -85,19 +87,25 @@ export async function createPidScopedAppLogProcess(
   }).catch(() => {
     state = 'failed';
   });
+  const cleanupSteps = new Set([
+    async () => {
+      stopped = true;
+      await active?.terminate();
+    },
+    async () => await monitor,
+    async () => await output[Symbol.asyncDispose](),
+  ]);
   let finishPromise: ReturnType<typeof finishPidScopedProcess> | undefined;
   const finish = async () =>
     (finishPromise ??= finishPidScopedProcess({
       options,
-      monitor,
-      active: () => active,
-      stop: () => {
-        stopped = true;
-      },
+      cleanupSteps,
       setState: (next) => {
         state = next;
       },
-      output,
+    }).then((outcome) => {
+      if (outcome.status === 'cleanup-pending') finishPromise = undefined;
+      return outcome;
     }));
   const startedAt = options.host.clock.now();
   return createAppLogLiveHandleFromFinish({
@@ -109,19 +117,11 @@ export async function createPidScopedAppLogProcess(
 async function finishPidScopedProcess(
   input: Readonly<{
     options: PidScopedAppLogProcessOptions;
-    monitor: Promise<void>;
-    active(): AppLogBackgroundProcess | undefined;
-    stop(): void;
+    cleanupSteps: Set<() => Promise<void>>;
     setState(state: AppLogLiveSnapshot['state']): void;
-    output: Awaited<ReturnType<AppLogRuntimeHost['outputs']['openAppend']>>;
   }>,
 ) {
-  input.stop();
-  const failures = await settleCleanupSteps([
-    async () => await input.active()?.terminate(),
-    async () => await input.monitor,
-    async () => await input.output[Symbol.asyncDispose](),
-  ]);
+  const failures = await settleCleanupSteps(input.cleanupSteps);
   if (failures.length > 0) {
     input.setState('failed');
     return {
@@ -141,11 +141,12 @@ async function finishPidScopedProcess(
   } as const;
 }
 
-async function settleCleanupSteps(steps: readonly (() => Promise<void>)[]): Promise<unknown[]> {
+async function settleCleanupSteps(steps: Set<() => Promise<void>>): Promise<unknown[]> {
   const failures: unknown[] = [];
   for (const step of steps) {
     try {
       await step();
+      steps.delete(step);
     } catch (error) {
       failures.push(error);
     }

packages/capture-kit/src/audio-probe-runtime.test.ts

@@ -82,6 +82,48 @@ function fakeHost(overrides: Partial<HostSystemAudioCaptureHost> = {}): HostSyst
   });
 }
 
+test('a host audio handle retries failed termination and disposes only once after confirmation', async () => {
+  await withStatusDir(async (dir) => {
+    const statusPath = path.join(dir, 'audio-probe.json');
+    let attempts = 0;
+    let exit!: (result: HostCommandResult) => void;
+    const wait = new Promise<HostCommandResult>((resolve) => {
+      exit = resolve;
+    });
+    const host = fakeHost({
+      start: async () => {
+        await fs.writeFile(
+          statusPath,
+          JSON.stringify({ state: 'running', rmsDbfs: [], peakDbfs: [], sampleCount: 0 }),
+        );
+        return {
+          marker,
+          wait,
+          terminate: async () => {
+            attempts += 1;
+            if (attempts === 1) throw new Error('Termination unavailable');
+            exit({ stdout: '', stderr: '', exitCode: 0 });
+          },
+        };
+      },
+    });
+    const started = await startHostAudioProbe({
+      host,
+      device,
+      owner,
+      input: { sessionId: 's1', statusPath, durationMs: 1000, bucketMs: 500, fence },
+    });
+    const handle = started.pendingHandle.transfer();
+    await assert.rejects(
+      async () => await handle[Symbol.asyncDispose](),
+      /Termination unavailable/,
+    );
+    await handle[Symbol.asyncDispose]();
+    await handle[Symbol.asyncDispose]();
+    assert.equal(attempts, 2);
+  });
+});
+
 test('descriptor codec round-trips and rejects malformed bodies', () => {
   const encoded = encodeDurableDescriptor(hostAudioProbeDescriptorCodec, descriptorBody);
   const decoded = hostAudioProbeDescriptorCodec.decode(encoded.body);

packages/capture-kit/src/audio-probe-runtime.ts

@@ -16,6 +16,7 @@ import type {
 } from '@agent-device/contracts/audio-probe-runtime-host';
 import {
   isConfirmedCleanup,
+  createRetryableCleanup,
   type CleanupOutcome,
   type FinishOutcome,
   type ReattachOutcome,
@@ -111,8 +112,6 @@ function createHostAudioProbeLiveHandle(
   process: HostAudioCaptureProcess,
 ): AudioProbeLiveHandle {
   let finish: Promise<FinishOutcome<AudioProbeCompletion>> | undefined;
-  let cleanup: Promise<CleanupOutcome> | undefined;
-  let disposal: Promise<void> | undefined;
   // The terminal result must stay observed: without it, a helper that dies after publishing a
   // `running` checkpoint would read as running forever and stop would fabricate a normal
   // completion. Only a terminal `stopped` publication outranks an observed exit — the helper
@@ -139,13 +138,13 @@ function createHostAudioProbeLiveHandle(
         result: finalizeStatus(snapshot, beforeStop, 'stopped'),
       } as const;
     })());
-  const forceCleanup = () =>
-    (cleanup ??= finish
+  const clean = () =>
+    finish
       ? finish.then(
           async () => ({ status: 'cleaned' }) as const,
           async () => await terminateForCleanup(process),
         )
-      : terminateForCleanup(process));
+      : terminateForCleanup(process);
   return Object.freeze({
     inspect: () => snapshot,
     status: async () => {
@@ -155,11 +154,7 @@ function createHostAudioProbeLiveHandle(
       return status ?? finalizeStatus(snapshot, undefined, 'not-started');
     },
     finish: finishProbe,
-    forceCleanup,
-    [Symbol.asyncDispose]: async () => {
-      disposal ??= forceCleanup().then(assertConfirmedAudioProbeCleanup);
-      await disposal;
-    },
+    ...createRetryableCleanup(clean, assertConfirmedAudioProbeCleanup),
   });
 }
 

packages/capture-kit/src/capture-admission/__tests__/screen-recording-session-resource.test.ts

@@ -1,10 +1,13 @@
 import { makeCaptureSessionBinding } from '../../durable-capture/session-binding.fixtures.ts';
+import fs from 'node:fs';
+import path from 'node:path';
 import { expect, test, vi } from 'vitest';
 import { PendingTransferGuard } from '@agent-device/contracts/async-lifecycle';
 import { localRuntimeOwner } from '@agent-device/contracts/platform-runtime';
 import type { ScreenRecordingLiveHandle } from '@agent-device/contracts/screen-recording-runtime';
 import { createDurableResourceEnvelope } from '../../durable-resource-envelope.ts';
 import { createScreenRecordingLiveHandle } from '../../screen-recording-live-handle.ts';
+import { mkdtempForTestSync } from '../../tmp-dir.fixtures.ts';
 import { stopAndExportScreenRecording } from '../../recording/stop-sequence.ts';
 import { createScreenRecordingAdmissionLedger } from '../screen-recording-admission-ledger.ts';
 import {
@@ -273,6 +276,124 @@ test('a record stop that fails after collecting resumes through the fence withou
   expect(sessionStore.get(sessionName)?.screenRecording).toBeUndefined();
 });
 
+/**
+ * What session teardown does with a recording `record stop` already committed: the queued disposal had
+ * read the session's recording before that stop, so it still holds the handle and the fence the record
+ * names. It is answered with the export the stop produced and signals the recorder no second time.
+ * Before the record could answer for its own export, this refusal surfaced as an unreleased resource on
+ * session cleanup, and a second `record stop` looked like a failure to callers.
+ */
+test('a disposal queued behind a committed recording stop replays that export', async () => {
+  const recording = await adoptCommittedRecording('screen-recording-replayed-stop-');
+  const stopped = await finishLiveScreenRecording({
+    intent: 'capture',
+    binding: recording.binding,
+  });
+  recording.requeue();
+
+  await expect(
+    finishLiveScreenRecording({ intent: 'disposal', binding: recording.binding }),
+  ).resolves.toEqual(stopped);
+  expect(recording.finish).toHaveBeenCalledOnce();
+  expect(recording.signals).toHaveBeenCalledOnce();
+  expect(recording.binding.read()).toBeUndefined();
+});
+
+/**
+ * A committed record whose export file no longer exists has nothing serveable, and this kind never
+ * hands a caller a path it cannot download. The surviving-file check belongs to the reader both
+ * consumers share, so a disposal queued behind a stop whose clip was removed keeps the typed
+ * refusal teardown settles, rather than reporting a successful stop for a deleted file.
+ */
+test('a disposal queued behind a deleted export refuses instead of serving the missing path', async () => {
+  const recording = await adoptCommittedRecording('screen-recording-deleted-stop-');
+  await finishLiveScreenRecording({ intent: 'capture', binding: recording.binding });
+  fs.rmSync(recording.outPath);
+  recording.requeue();
+
+  await expect(
+    finishLiveScreenRecording({ intent: 'disposal', binding: recording.binding }),
+  ).rejects.toMatchObject({ details: { reason: 'resource-already-completed' } });
+  expect(recording.signals).toHaveBeenCalledOnce();
+});
+
+/**
+ * A recording adopted into a session, with the recorder's stop driven through the real export
+ * sequence, so the committed record holds the stop response the kind journals on commit.
+ */
+async function adoptCommittedRecording(prefix: string) {
+  const sessionStore = makeCaptureAdmissionSessionStore<TestRecordingSession>(prefix);
+  const sessionName = 'recording';
+  const session: TestRecordingSession = { name: sessionName, device };
+  sessionStore.set(sessionName, session);
+  const binding = makeCaptureSessionBinding(sessionStore, sessionName, {
+    read: (session) => session.screenRecording,
+    replace: (session, screenRecording) => ({ ...session, screenRecording }),
+  });
+  const owner = localRuntimeOwner('android');
+  const fence = { token: 'recording-fence', generation: 1 } as const;
+  const outPath = path.join(mkdtempForTestSync(`${prefix}export-`), 'recording.mp4');
+  fs.writeFileSync(outPath, 'mp4');
+  const signals = vi.fn(async () => ({ observation: { recorder: 'confirmed' as const } }));
+  const handle = createScreenRecordingLiveHandle(
+    {
+      backend: 'android',
+      outPath,
+      startedAt: 1,
+      scope: 'app',
+      showTouches: false,
+      recordOnlySession: false,
+      gestureEvents: [],
+    },
+    {
+      finish: (snapshot, progress) =>
+        stopAndExportScreenRecording({
+          snapshot,
+          progress,
+          steps: {
+            stop: signals,
+            collect: async () => {},
+            finalize: async () => ({ nativePathDisposition: 'retired' as const }),
+            discard: async () => {},
+          },
+        }),
+      forceCleanup: async () => ({ status: 'cleaned' }),
+    },
+  );
+  const finish = vi.fn(handle.finish);
+  await adoptStartedScreenRecording({
+    admissionLedger: createScreenRecordingAdmissionLedger(),
+    binding,
+    device: session.device,
+    owner,
+    fence,
+    pendingHandle: new PendingTransferGuard({ ...handle, finish }),
+    envelope: createDurableResourceEnvelope({
+      resourceKind: 'screen-recording',
+      sessionId: sessionName,
+      device: { id: session.device.id, family: 'android', kind: 'emulator' },
+      owner,
+      fence,
+      lifecycle: 'open',
+      descriptor: { version: 1, body: { recordingId: 'recording-id' } },
+    }),
+    throwIfCanceled: () => {},
+  });
+  // The queued teardown's own view of the slot, captured while the record was still open.
+  const queued = binding.read();
+  if (!queued) throw new Error('Expected adopted screen recording');
+  return {
+    binding,
+    finish,
+    signals,
+    outPath,
+    /** Puts the slot back the way the queued teardown left it holding it. */
+    requeue: () => {
+      binding.adopt(queued);
+    },
+  };
+}
+
 function resourcePath(
   sessionStore: CaptureAdmissionSessionStore<TestRecordingSession>,
   sessionName: string,