diff --git a/packages/capture-kit/src/app-log-live-handle.ts b/packages/capture-kit/src/app-log-live-handle.ts index 0e7e2f252d..a2c4d9c35d 100644 --- a/packages/capture-kit/src/app-log-live-handle.ts +++ b/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> | undefined; - let cleanup: Promise | undefined; - let disposal: Promise | undefined; - const forceCleanup = (): Promise => (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), }); } diff --git a/packages/capture-kit/src/app-log-pid-process.test.ts b/packages/capture-kit/src/app-log-pid-process.test.ts index adc52db003..22b91dfe9d 100644 --- a/packages/capture-kit/src/app-log-pid-process.test.ts +++ b/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 }>(); diff --git a/packages/capture-kit/src/app-log-pid-process.ts b/packages/capture-kit/src/app-log-pid-process.ts index cdbf14f31c..7e9accd46a 100644 --- a/packages/capture-kit/src/app-log-pid-process.ts +++ b/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 | 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; - active(): AppLogBackgroundProcess | undefined; - stop(): void; + cleanupSteps: Set<() => Promise>; setState(state: AppLogLiveSnapshot['state']): void; - output: Awaited>; }>, ) { - 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)[]): Promise { +async function settleCleanupSteps(steps: Set<() => Promise>): Promise { const failures: unknown[] = []; for (const step of steps) { try { await step(); + steps.delete(step); } catch (error) { failures.push(error); } diff --git a/packages/capture-kit/src/audio-probe-runtime.test.ts b/packages/capture-kit/src/audio-probe-runtime.test.ts index dc121959aa..9fe0e2a558 100644 --- a/packages/capture-kit/src/audio-probe-runtime.test.ts +++ b/packages/capture-kit/src/audio-probe-runtime.test.ts @@ -82,6 +82,48 @@ function fakeHost(overrides: Partial = {}): 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((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); diff --git a/packages/capture-kit/src/audio-probe-runtime.ts b/packages/capture-kit/src/audio-probe-runtime.ts index 4d323f90e5..6ec6353313 100644 --- a/packages/capture-kit/src/audio-probe-runtime.ts +++ b/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> | undefined; - let cleanup: Promise | undefined; - let disposal: Promise | 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), }); } diff --git a/packages/capture-kit/src/capture-admission/__tests__/screen-recording-session-resource.test.ts b/packages/capture-kit/src/capture-admission/__tests__/screen-recording-session-resource.test.ts index 99e3b36278..86bec99fe1 100644 --- a/packages/capture-kit/src/capture-admission/__tests__/screen-recording-session-resource.test.ts +++ b/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(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, sessionName: string, diff --git a/packages/capture-kit/src/capture-admission/app-log-durable-resource-replay.test.ts b/packages/capture-kit/src/capture-admission/app-log-durable-resource-replay.test.ts new file mode 100644 index 0000000000..0c76a99167 --- /dev/null +++ b/packages/capture-kit/src/capture-admission/app-log-durable-resource-replay.test.ts @@ -0,0 +1,198 @@ +import { expect, test, vi } from 'vitest'; +import type { AppLogCompletion, AppLogLiveHandle } from '@agent-device/contracts/app-log-runtime'; +import { PendingTransferGuard } from '@agent-device/contracts/async-lifecycle'; +import { localRuntimeOwner } from '@agent-device/contracts/platform-runtime'; +import type { JsonObject } from '@agent-device/contracts/client'; +import type { DeviceInfo } from '@agent-device/kernel/device'; +import { createDurableResourceEnvelope } from '../durable-resource-envelope.ts'; +import { makeCaptureSessionBinding } from '../durable-capture/session-binding.fixtures.ts'; +import { appLogDurableResource, appLogResourceStore } from './app-log-durable-resource.ts'; +import { createDurableCaptureAdmissionLedger } from './durable-capture-admission-ledger.ts'; +import type { DurableCaptureSessionResource } from '../durable-capture/index.ts'; +import { + makeCaptureAdmissionSessionStore, + type CaptureAdmissionSessionStore, +} from './__tests__/session-store.fixtures.ts'; + +const device: DeviceInfo = { + platform: 'android', + id: 'emulator-5554', + name: 'Pixel', + kind: 'emulator', +}; + +const completion: AppLogCompletion = { + backend: 'android', + outputPath: '/tmp/app.log', + completedAt: 2, +}; + +/** The app-log slot this kind owns, plus the device the record answers the caller with. */ +type TestAppLogSession = Readonly<{ + name: string; + device: DeviceInfo; + appLog?: DurableCaptureSessionResource<'app-log', AppLogLiveHandle>; +}>; + +function makeStore(prefix: string) { + const sessionStore = makeCaptureAdmissionSessionStore(prefix); + sessionStore.set('logging', { name: 'logging', device }); + return sessionStore; +} + +function makeBinding(sessionStore: CaptureAdmissionSessionStore) { + return makeCaptureSessionBinding<'app-log', AppLogLiveHandle, TestAppLogSession>( + sessionStore, + 'logging', + { + read: (session) => session.appLog, + replace: (session, appLog) => ({ ...session, appLog }), + }, + ); +} + +function makeHandle(finishImpl?: () => Promise) { + return { + inspect: () => ({ backend: 'android', state: 'active' as const, startedAt: 1 }), + finish: vi.fn(async () => { + await finishImpl?.(); + return { status: 'completed' as const, result: completion }; + }), + forceCleanup: vi.fn(async () => ({ status: 'cleaned' as const })), + [Symbol.asyncDispose]: async () => {}, + } satisfies AppLogLiveHandle & { finish: ReturnType }; +} + +async function adopt( + sessionStore: CaptureAdmissionSessionStore, + handle: AppLogLiveHandle, +) { + const binding = makeBinding(sessionStore); + await appLogDurableResource.adoptStarted({ + admissionLedger: createDurableCaptureAdmissionLedger({ displayName: 'app-log' }), + binding, + device, + owner: localRuntimeOwner(device.platform), + fence: { token: 'app-log-fence', generation: 1 }, + pendingHandle: new PendingTransferGuard(handle), + envelope: createDurableResourceEnvelope({ + resourceKind: 'app-log', + sessionId: 'logging', + device: { id: device.id, family: device.platform, kind: device.kind }, + owner: localRuntimeOwner(device.platform), + fence: { token: 'app-log-fence', generation: 1 }, + lifecycle: 'open', + descriptor: { version: 1, body: { pid: 123 } }, + }), + throwIfCanceled: () => {}, + }); + return binding; +} + +function resourcePath(sessionStore: CaptureAdmissionSessionStore) { + return appLogResourceStore.resolvePath(sessionStore.resolveSessionDir('logging')); +} + +/** + * Commits a record the way the mechanics do, then overwrites its journal with `metadata`, so a + * finish arrives at a committed record whose completion facts are damaged on disk. + */ +async function commitRecordWithMetadata( + sessionStore: CaptureAdmissionSessionStore, + metadata: JsonObject | undefined, +) { + const handle = makeHandle(); + const binding = await adopt(sessionStore, handle); + const queued = binding.read(); + if (!queued) throw new Error('Expected an adopted app-log resource'); + await appLogDurableResource.forceCleanupLive({ binding }); + const record = appLogResourceStore.read(resourcePath(sessionStore)); + if (record.status !== 'decoded') throw new Error('Expected a committed app-log record'); + appLogResourceStore.write(resourcePath(sessionStore), { ...record.envelope, metadata }); + // A finish that had captured the handle before the record settled. + binding.adopt(queued); + return { binding, handle }; +} + +/** + * A stopped stream journals the three facts it answered with, so a second caller of the same record + * is answered from them instead of being refused. Before the record could answer for its own + * completion, a teardown finish queued behind an in-flight `logs stop` reported the session's log + * file as an unreleased resource. + */ +test('a finish queued behind a committed app-log stop replays that completion', async () => { + const sessionStore = makeStore('app-log-replay-'); + const handle = makeHandle(); + const binding = await adopt(sessionStore, handle); + // 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 an adopted app-log resource'); + + const stopped = await appLogDurableResource.finishLive({ intent: 'capture', binding }); + binding.adopt(queued); + + await expect(appLogDurableResource.finishLive({ intent: 'disposal', binding })).resolves.toEqual( + stopped, + ); + expect(handle.finish).toHaveBeenCalledOnce(); + expect(binding.read()).toBeUndefined(); +}); + +/** + * A stream the session forced closed committed a record that never produced a log file, so its + * journal holds the mechanics' cleanup state and none of the completion facts. A finish arriving for + * that record keeps the typed refusal: nothing here may invent a log path the stream never flushed to. + */ +test('an app-log record committed by cleanup refuses rather than inventing a log path', async () => { + const sessionStore = makeStore('app-log-replay-cleanup-'); + const handle = makeHandle(); + const binding = await adopt(sessionStore, handle); + const queued = binding.read(); + if (!queued) throw new Error('Expected an adopted app-log resource'); + + await appLogDurableResource.forceCleanupLive({ binding }); + // A finish that had already captured this handle reaches the record cleanup committed. + binding.adopt(queued); + + await expect( + appLogDurableResource.finishLive({ intent: 'disposal', binding }), + ).rejects.toMatchObject({ details: { reason: 'resource-already-completed' } }); + expect(handle.finish).not.toHaveBeenCalled(); +}); + +/** + * Every fact the reader needs is checked, so a committed record whose journal is damaged stays a + * refusal. Each case is a record that has lost exactly one fact, or holds one the backend union does + * not know: none of them may be answered with a completion assembled from the rest. + */ +const damagedCompletions: ReadonlyArray = [ + ['no journal at all', undefined], + ['no backend', { phase: 'completed', outputPath: '/tmp/app.log', completedAt: 2 }] as const, + [ + 'an unknown backend', + { phase: 'completed', backend: 'symbian', outputPath: '/tmp/app.log', completedAt: 2 }, + ] as const, + ['no log path', { phase: 'completed', backend: 'android', completedAt: 2 }] as const, + [ + 'an empty log path', + { phase: 'completed', backend: 'android', outputPath: '', completedAt: 2 }, + ] as const, + ['no stop time', { phase: 'completed', backend: 'android', outputPath: '/tmp/app.log' }] as const, + [ + 'a stop time that is not a real instant', + { phase: 'completed', backend: 'android', outputPath: '/tmp/app.log', completedAt: Number.NaN }, + ] as const, +]; + +test.each(damagedCompletions)( + 'a committed app-log record that cannot name its completion (%s) is refused', + async (_case, metadata) => { + const sessionStore = makeStore('app-log-replay-damaged-'); + const { binding, handle } = await commitRecordWithMetadata(sessionStore, metadata); + + await expect( + appLogDurableResource.finishLive({ intent: 'disposal', binding }), + ).rejects.toMatchObject({ details: { reason: 'resource-already-completed' } }); + expect(handle.finish).not.toHaveBeenCalled(); + }, +); diff --git a/packages/capture-kit/src/capture-admission/app-log-durable-resource.ts b/packages/capture-kit/src/capture-admission/app-log-durable-resource.ts index 6383f886a6..12366c5c82 100644 --- a/packages/capture-kit/src/capture-admission/app-log-durable-resource.ts +++ b/packages/capture-kit/src/capture-admission/app-log-durable-resource.ts @@ -1,7 +1,45 @@ import type { AppLogCompletion, AppLogLiveHandle } from '@agent-device/contracts/app-log-runtime'; +import type { JsonObject } from '@agent-device/contracts/client'; import { createDurableCaptureResource } from './durable-capture-resource.ts'; import { appLogResourceStore } from './app-log-resource-store.ts'; +const LOG_BACKENDS: readonly AppLogCompletion['backend'][] = [ + 'ios-simulator', + 'ios-device', + 'android', + 'harmonyos', + 'macos', +]; + +/** + * The stopped stream answered with the log path it flushed to, its backend, and when it stopped — and + * those are exactly the facts this kind journals on commit, so a caller that reaches the committed + * record can be answered from them. Anything else on the record is the mechanics' own state, which is + * not a completion: a record that cannot name all three is refused rather than answered. + */ +function readAppLogStoredCompletion( + metadata: JsonObject | undefined, +): AppLogCompletion | undefined { + const backend = metadata?.backend; + const outputPath = metadata?.outputPath; + const completedAt = metadata?.completedAt; + if ( + typeof backend !== 'string' || + !LOG_BACKENDS.includes(backend as AppLogCompletion['backend']) || + typeof outputPath !== 'string' || + outputPath.length === 0 || + typeof completedAt !== 'number' || + !Number.isFinite(completedAt) + ) { + return undefined; + } + return { + backend: backend as AppLogCompletion['backend'], + outputPath, + completedAt, + }; +} + export const appLogDurableResource = createDurableCaptureResource< 'app-log', AppLogLiveHandle, @@ -15,6 +53,7 @@ export const appLogDurableResource = createDurableCaptureResource< outputPath: completion.outputPath, completedAt: completion.completedAt, }), + completionFromMetadata: readAppLogStoredCompletion, failedFinishPolicy: 'dispose-on-failed-finish', messages: { noActive: 'no app log stream active', diff --git a/packages/capture-kit/src/capture-admission/screen-recording-resource-store.ts b/packages/capture-kit/src/capture-admission/screen-recording-resource-store.ts index 6010c92836..a98ac339d3 100644 --- a/packages/capture-kit/src/capture-admission/screen-recording-resource-store.ts +++ b/packages/capture-kit/src/capture-admission/screen-recording-resource-store.ts @@ -6,3 +6,10 @@ export const screenRecordingResourceStore = createDurableCaptureResourceStore({ fileName: 'screen-recording.resource.json', displayName: 'screen recording', }); + +/** + * The metadata key this record stores a finished recording's stop response under, beside the + * mechanics' own facts. The writer is the kind's `completionMetadata`, the reader is + * `screen-recording-stop-recovery.ts`; both name this key so a replay cannot miss what was written. + */ +export const SCREEN_RECORDING_COMPLETION_METADATA_KEY = 'completion'; diff --git a/packages/capture-kit/src/capture-admission/screen-recording-session-resource.ts b/packages/capture-kit/src/capture-admission/screen-recording-session-resource.ts index 9a477dfbaf..a5ff0a579b 100644 --- a/packages/capture-kit/src/capture-admission/screen-recording-session-resource.ts +++ b/packages/capture-kit/src/capture-admission/screen-recording-session-resource.ts @@ -21,7 +21,10 @@ import type { import { createDurableCaptureResource } from './durable-capture-resource.ts'; import type { DurableCaptureFinishIntent } from './durable-capture-resource.ts'; import type { ScreenRecordingAdmissionLedger } from './screen-recording-admission-ledger.ts'; -import { screenRecordingResourceStore } from './screen-recording-resource-store.ts'; +import { + SCREEN_RECORDING_COMPLETION_METADATA_KEY, + screenRecordingResourceStore, +} from './screen-recording-resource-store.ts'; export const screenRecordingDurableResource = createDurableCaptureResource< 'screen-recording', @@ -32,6 +35,12 @@ export const screenRecordingDurableResource = createDurableCaptureResource< displayName: 'screen recording', store: screenRecordingResourceStore, completionMetadata: encodeScreenRecordingCompletionMetadata, + // The manifest holds the whole stop response this kind wrote, so a caller that reaches the committed + // record is served the same export the first stop was answered with. That answer is a cold path: the + // record has to be read before anyone can be refused or served, so the reader is reached on demand + // rather than added to what importing this kind evaluates. + completionFromMetadata: async (metadata) => + (await import('./screen-recording-stop-recovery.ts')).readReplayableStoredCompletion(metadata), // ADR 0024 rule 6: the next stop re-collects the native artifact a failed export left behind, and // forced cleanup would delete exactly that. Disposal belongs to teardown and start rollback. failedFinishPolicy: 'preserve-retry-material', @@ -62,13 +71,6 @@ export function finishLiveScreenRecording(params: { return screenRecordingDurableResource.finishLive(params); } -/** - * The manifest key holding a finished recording's stop response. The completion is stored as the one - * object `record stop` returned, so a replay cannot lose a field on its way through the manifest; - * `screen-recording-stop-recovery.ts` reads it back and serves it. - */ -export const SCREEN_RECORDING_COMPLETION_METADATA_KEY = 'completion'; - export function encodeScreenRecordingCompletionMetadata( completion: ScreenRecordingCompletion, ): JsonObject { diff --git a/packages/capture-kit/src/capture-admission/screen-recording-stop-recovery.ts b/packages/capture-kit/src/capture-admission/screen-recording-stop-recovery.ts index 74fec4dc93..092f6eb8b6 100644 --- a/packages/capture-kit/src/capture-admission/screen-recording-stop-recovery.ts +++ b/packages/capture-kit/src/capture-admission/screen-recording-stop-recovery.ts @@ -20,8 +20,8 @@ import { emitDiagnostic } from '@agent-device/host-kit/diagnostics'; import type { DurableCaptureResourceRecord } from '../durable-capture/index.ts'; import { SCREEN_RECORDING_COMPLETION_METADATA_KEY, - screenRecordingDurableResource, -} from './screen-recording-session-resource.ts'; + screenRecordingResourceStore, +} from './screen-recording-resource-store.ts'; /** * What a `record stop` owes a session, decided from the durable recording manifest alone. @@ -56,7 +56,7 @@ export function resolveScreenRecordingStopRecovery( if (record.status !== 'decoded') return { kind: 'none' }; assertManifestBelongsToRequest(params, record.envelope); if (record.envelope.lifecycle !== 'completed') return { kind: 'open', resourcePath }; - const completion = readStoredCompletion(record.envelope.metadata); + const completion = readReplayableStoredCompletion(record.envelope.metadata); if (completion === undefined) { emitDiagnostic({ level: 'warn', @@ -65,7 +65,6 @@ export function resolveScreenRecordingStopRecovery( }); return { kind: 'none' }; } - if (!fs.existsSync(completion.outPath)) return { kind: 'none' }; return { kind: 'completed', completion }; } @@ -94,11 +93,10 @@ function readSessionManifest(params: ScreenRecordingManifestParams): Readonly<{ resourcePath: string; record: DurableCaptureResourceRecord<'screen-recording'>; }> { - const resourcePath = screenRecordingDurableResource.resourcePath( - params.sessionStore, - params.sessionName, + const resourcePath = screenRecordingResourceStore.resolvePath( + params.sessionStore.resolveSessionDir(params.sessionName), ); - return { resourcePath, record: screenRecordingDurableResource.store.read(resourcePath) }; + return { resourcePath, record: screenRecordingResourceStore.read(resourcePath) }; } function assertManifestBelongsToRequest( @@ -122,12 +120,17 @@ function assertManifestBelongsToRequest( } /** - * The stop response the manifest stored, or nothing when it is not a whole one. The envelope store - * has already bounded this metadata to frozen plain JSON, so what recovery still has to establish is - * that the stored response answers correctly: the path it serves, the caller-side paths a remote + * The stop response the manifest stored, or nothing when it cannot be served. The envelope store + * has already bounded this metadata to frozen plain JSON, so what this reader still has to establish + * is that the stored response answers correctly: the path it serves, the caller-side paths a remote * download is named after, and the duration the response subtracts. + * + * It owns the surviving-file check on top of that, because every caller is answered with the path the + * record names: two reach this reader — a stop whose response never reached its caller, and a finish + * that reached the record after another caller committed it — and neither may hand over a file that is + * gone. A caller never receives a path it cannot download. */ -function readStoredCompletion( +export function readReplayableStoredCompletion( metadata: JsonObject | undefined, ): ScreenRecordingCompletion | undefined { const stored = metadata?.[SCREEN_RECORDING_COMPLETION_METADATA_KEY]; @@ -139,7 +142,8 @@ function readStoredCompletion( ) { return undefined; } - return stored as unknown as ScreenRecordingCompletion; + const completion = stored as unknown as ScreenRecordingCompletion; + return fs.existsSync(completion.outPath) ? completion : undefined; } /** The values a stop response computes on rather than repeats. */ diff --git a/packages/capture-kit/src/durable-capture/definition.ts b/packages/capture-kit/src/durable-capture/definition.ts index 0b947fdae9..f9adebf12a 100644 --- a/packages/capture-kit/src/durable-capture/definition.ts +++ b/packages/capture-kit/src/durable-capture/definition.ts @@ -57,6 +57,19 @@ export type DurableCaptureRecordDefinition = Readonly<{ displayName: string; store: DurableCaptureResourceStore; completionMetadata(result: C): JsonObject; + /** + * The completion this kind stored in a record that is already committed, or nothing it can serve. + * + * A committed record is the only account of an outcome a queued caller has never seen, so the kind + * that wrote it is the one that reads it back: that caller is then answered with the outcome the + * first finish produced instead of being refused, or driven through work that already happened. A + * reader answers `undefined` for a stored shape it does not recognise, so a partial or foreign record + * cannot yield a plausible-looking outcome. + * + * Optional because a kind whose stored metadata cannot carry its whole outcome has nothing honest to + * replay; its callers then get the terminal refusal. + */ + completionFromMetadata?(metadata: JsonObject | undefined): C | undefined | Promise; failedFinishPolicy: DurableCaptureFailedFinishPolicy; messages: Readonly<{ noActive: string; diff --git a/packages/capture-kit/src/durable-capture/durable-capture.fixtures.ts b/packages/capture-kit/src/durable-capture/durable-capture.fixtures.ts index 9496e6845f..095c756b12 100644 --- a/packages/capture-kit/src/durable-capture/durable-capture.fixtures.ts +++ b/packages/capture-kit/src/durable-capture/durable-capture.fixtures.ts @@ -28,6 +28,9 @@ export const TEST_CAPTURE_KIND = 'test-capture'; export type TestCaptureCompletion = Readonly<{ outputPath: string; completedAt: number }>; +/** Where this test kind stores its completion so a later caller of the same record can be answered. */ +export const TEST_CAPTURE_COMPLETION_METADATA_KEY = 'completion'; + export type TestCaptureHandle = AsyncDisposable & Readonly<{ finish(): Promise>; @@ -56,9 +59,24 @@ export function createTestCaptureDefinition( store, failedFinishPolicy, completionMetadata: (completion) => ({ - outputPath: completion.outputPath, - completedAt: completion.completedAt, + [TEST_CAPTURE_COMPLETION_METADATA_KEY]: { + outputPath: completion.outputPath, + completedAt: completion.completedAt, + }, }), + completionFromMetadata: (metadata) => { + const stored = metadata?.[TEST_CAPTURE_COMPLETION_METADATA_KEY]; + if ( + typeof stored !== 'object' || + stored === null || + Array.isArray(stored) || + typeof (stored as Record).outputPath !== 'string' || + typeof (stored as Record).completedAt !== 'number' + ) { + return undefined; + } + return stored as unknown as TestCaptureCompletion; + }, messages: { noActive: 'no test capture active', cleanupPendingHint: 'Keep the test capture manifest for exact-owner recovery.', diff --git a/packages/capture-kit/src/durable-capture/fence.test.ts b/packages/capture-kit/src/durable-capture/fence.test.ts index 81c753cd00..acc0ffe112 100644 --- a/packages/capture-kit/src/durable-capture/fence.test.ts +++ b/packages/capture-kit/src/durable-capture/fence.test.ts @@ -3,7 +3,10 @@ import { expect, test, vi } from 'vitest'; import { localRuntimeOwner } from '@agent-device/contracts/platform-runtime'; import { createDurableResourceEnvelope } from '../durable-resource-envelope.ts'; import { mkdtempForTestSync } from '../tmp-dir.fixtures.ts'; -import { withDurableCaptureResourceFence } from './fence.ts'; +import { + withDurableCaptureResourceFence, + type DurableCaptureOpenResourceFenceLease, +} from './fence.ts'; import { createDurableCaptureResourceStore } from './store.ts'; const store = createDurableCaptureResourceStore({ @@ -20,6 +23,7 @@ test('rejects a stale fence before its side effect', async () => { store, resourcePath, expected: { token: 'stale', generation: 1 }, + onCompleted: () => undefined, run: sideEffect, }), ).rejects.toMatchObject({ details: { reason: 'ownership-fence-lost' } }); @@ -41,6 +45,7 @@ test('serializes validation, native work, and transition for one resource path', store, resourcePath, expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, run: async (lease) => { order.push('first-start'); started(); @@ -53,6 +58,7 @@ test('serializes validation, native work, and transition for one resource path', store, resourcePath, expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, run: async () => { order.push('second'); }, @@ -68,6 +74,168 @@ test('serializes validation, native work, and transition for one resource path', }); }); +/** + * A caller that captured the same live resource presents the same fence, so serialization alone lets + * it drive a record its predecessor already committed. Both facts here have to hold: the caller that + * arrives second is answered without a lease, and the record stays as its predecessor left it. + */ +test('a caller holding the same fence cannot write a record an earlier caller committed', async () => { + const resourcePath = makeRecord(); + let release!: () => void; + let started!: () => void; + const firstStarted = new Promise((resolve) => { + started = resolve; + }); + const firstHeld = new Promise((resolve) => { + release = resolve; + }); + const first = withDurableCaptureResourceFence({ + store, + resourcePath, + expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, + run: async (lease) => { + lease.transition('completed', { metadata: { phase: 'completed' } }); + started(); + // Holds the fence, and the record its run committed, until the second caller is queued. + await firstHeld; + }, + }); + const late = vi.fn(async () => undefined); + const second = withDurableCaptureResourceFence({ + store, + resourcePath, + expected: { token: 'current', generation: 1 }, + onCompleted: (lease) => lease.envelope.metadata?.phase, + run: late, + }); + await firstStarted; + release(); + expect(await second).toBe('completed'); + await first; + expect(late).not.toHaveBeenCalled(); + expect(store.read(resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); + +/** + * A released lease must not publish while a later run holds the still-open record. The later run's + * own lease must remain usable after that refusal and commit the record successfully. + */ +test('a lease released with its run cannot write an open record held by a later run', async () => { + const resourcePath = makeRecord(); + let escape!: DurableCaptureOpenResourceFenceLease<'screen-recording'>; + await withDurableCaptureResourceFence({ + store, + resourcePath, + expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, + run: async (lease) => { + escape = lease; + }, + }); + + await withDurableCaptureResourceFence({ + store, + resourcePath, + expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, + run: async (lease) => { + const before = store.read(resourcePath); + expect(before).toMatchObject({ status: 'decoded', envelope: { lifecycle: 'open' } }); + let escapedError: unknown; + try { + escape.transition('open', { metadata: { phase: 'escaped-write' } }); + } catch (error) { + escapedError = error; + } + expect(escapedError).toMatchObject({ + details: { reason: 'ownership-fence-lost' }, + message: 'Screen recording resource fence was released with its run', + }); + expect(store.read(resourcePath)).toEqual(before); + lease.transition('completed', { metadata: { phase: 'completed' } }); + }, + }); + + expect(() => escape.transition('open', { metadata: { phase: 'completing' } })).toThrow(); + expect(store.read(resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); + +/** + * The reviewer's probe: an escaped lease invoked while the committing run still holds the fence. Its + * cached `open` view must not publish, or the manifest that run just committed is overwritten. + */ +test('an escaped lease cannot overwrite the record the committing run is writing', async () => { + const resourcePath = makeRecord(); + let escape!: DurableCaptureOpenResourceFenceLease<'screen-recording'>; + await withDurableCaptureResourceFence({ + store, + resourcePath, + expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, + run: async (lease) => { + escape = lease; + }, + }); + + let escapeWrite: unknown; + await withDurableCaptureResourceFence({ + store, + resourcePath, + expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, + run: async (lease) => { + lease.transition('completed', { metadata: { phase: 'completed' } }); + // The escaped lease is invoked with the committed record already published. + try { + escape.transition('open', { metadata: { phase: 'completing' } }); + escapeWrite = 'published'; + } catch (error) { + escapeWrite = error; + } + }, + }); + + expect(escapeWrite).toMatchObject({ details: { reason: 'resource-already-completed' } }); + expect(store.read(resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); + +/** A stop's progress port writes into the record it holds. A commit ends what it may still write. */ +test('a progress write that reaches a committed record cannot reopen it', async () => { + const resourcePath = makeRecord(); + let late!: () => void; + const held = new Promise((resolve) => { + late = resolve; + }); + const first = withDurableCaptureResourceFence({ + store, + resourcePath, + expected: { token: 'current', generation: 1 }, + onCompleted: () => undefined, + run: async (lease) => { + lease.transition('completed', { metadata: { phase: 'completed' } }); + late(); + // The stop's own progress port, driven after its record already committed. + lease.transition('open', { metadata: { phase: 'completing' } }); + }, + }); + await held; + await expect(first).rejects.toMatchObject({ details: { reason: 'resource-already-completed' } }); + expect(store.read(resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); + function makeRecord(): string { const resourcePath = store.resolvePath( path.join(mkdtempForTestSync('capture-fence-'), 'session'), diff --git a/packages/capture-kit/src/durable-capture/fence.ts b/packages/capture-kit/src/durable-capture/fence.ts index 970849013c..dbb37585a9 100644 --- a/packages/capture-kit/src/durable-capture/fence.ts +++ b/packages/capture-kit/src/durable-capture/fence.ts @@ -4,11 +4,12 @@ import type { DurableResourceLifecycleState, } from '@agent-device/contracts/durable-resource-envelope'; import type { ResourceOwnershipFence } from '@agent-device/contracts/platform-runtime'; -import type { DurableCaptureResourceStore } from './store.ts'; +import type { DurableCaptureResourceRecord, DurableCaptureResourceStore } from './store.ts'; const resourceFenceTails = new Map>(); -export type DurableCaptureResourceFenceLease = Readonly<{ +/** The lease over a record this fence found open; it is the only way a run may write that record. */ +export type DurableCaptureOpenResourceFenceLease = Readonly<{ envelope: DurableResourceEnvelope; transition( lifecycle: DurableResourceLifecycleState, @@ -19,50 +20,97 @@ export type DurableCaptureResourceFenceLease = Readonly<{ ): DurableResourceEnvelope; }>; -/** Holds one persisted ownership fence from validation through side effect and transition. */ +/** The facts of a record this fence found committed. A terminal record is read, never written. */ +export type DurableCaptureCompletedResourceFenceLease = Readonly<{ + envelope: DurableResourceEnvelope; +}>; + +/** + * Holds one persisted ownership fence from validation through side effect and transition. + * + * Fence equality proves ownership, never admission: two callers that captured the same live resource + * present the same fence, so serialization alone would let a queued attempt drive a record another + * attempt already committed. This is the one owner of that rule. While it holds the fence it reads + * the record's lifecycle, and a `completed` record is terminal — a run that finds one never receives + * a lease, so it cannot reach native or destructive work, and answers from `onCompleted` instead. + * + * Admission is re-checked at every write, not only at lease issue. A run can hand its lease or its + * progress port to something that outlives it, so a transition re-reads the persisted record and + * publishes onto that: a lease whose own run has ended, one that finds the fence gone, and one that + * finds the record committed are each refused before publication. That is what keeps an escaped + * progress or completion write from overwriting the outcome another run committed. + */ export async function withDurableCaptureResourceFence( params: Readonly<{ store: DurableCaptureResourceStore; resourcePath: string; expected: ResourceOwnershipFence; - run(lease: DurableCaptureResourceFenceLease): Promise; + /** Answers a run whose record this fence found committed, before any of the run's work. */ + onCompleted(lease: DurableCaptureCompletedResourceFenceLease): Result | Promise; + run(lease: DurableCaptureOpenResourceFenceLease): Promise; }>, ): Promise { return await serializeResource(params.resourcePath, async () => { - const record = params.store.read(params.resourcePath); - if (record.status !== 'decoded') { - throw fenceError( - params.store, - record.status === 'missing' - ? `${params.store.displayName} resource record is missing` - : `${params.store.displayName} resource record is unreattachable: ${record.message}`, - ); + const admitted = requireAdmissibleRecord(params.store, params.resourcePath, params.expected); + if (admitted.envelope.lifecycle === 'completed') { + return await params.onCompleted({ envelope: admitted.envelope }); } - if (!sameFence(record.envelope.fence, params.expected)) { - throw fenceError( - params.store, - `${params.store.displayName} resource ownership fence was lost`, - ); + let current = admitted.envelope; + // The lease is the run's, for as long as that run is the one holding the fence. + let issued = true; + try { + return await params.run({ + get envelope() { + return current; + }, + transition(lifecycle, update = {}) { + // The persisted record decides, so a lease cannot publish from a view its own run cached. + const persisted = requireAdmissibleRecord( + params.store, + params.resourcePath, + params.expected, + ); + if (persisted.envelope.lifecycle === 'completed') { + throw completedDurableCaptureError(params.store); + } + if (!issued) throw releasedLeaseError(params.store); + current = Object.freeze({ + ...persisted.envelope, + lifecycle, + ...(update.descriptor === undefined ? {} : { descriptor: update.descriptor }), + ...(update.metadata === undefined ? {} : { metadata: update.metadata }), + }); + params.store.write(params.resourcePath, current); + return current; + }, + }); + } finally { + issued = false; } - let current = record.envelope; - return await params.run({ - get envelope() { - return current; - }, - transition(lifecycle, update = {}) { - current = Object.freeze({ - ...current, - lifecycle, - ...(update.descriptor === undefined ? {} : { descriptor: update.descriptor }), - ...(update.metadata === undefined ? {} : { metadata: update.metadata }), - }); - params.store.write(params.resourcePath, current); - return current; - }, - }); }); } +/** The record this caller still owns, refusing a record it no longer holds a fence over. */ +function requireAdmissibleRecord( + store: DurableCaptureResourceStore, + resourcePath: string, + expected: ResourceOwnershipFence, +): Extract, { status: 'decoded' }> { + const record = store.read(resourcePath); + if (record.status !== 'decoded') { + throw fenceError( + store, + record.status === 'missing' + ? `${store.displayName} resource record is missing` + : `${store.displayName} resource record is unreattachable: ${record.message}`, + ); + } + if (!sameFence(record.envelope.fence, expected)) { + throw fenceError(store, `${store.displayName} resource ownership fence was lost`); + } + return record; +} + function sameFence(left: ResourceOwnershipFence, right: ResourceOwnershipFence): boolean { return left.token === right.token && left.generation === right.generation; } @@ -94,3 +142,36 @@ function fenceError(store: DurableCaptureResourceStore, message: string) hint: `Use the current ${store.displayName.toLowerCase()} resource owner or retain the recovery record for manual recovery.`, }); } + +/** + * The refusal a lease gives once the run it was issued to has ended. The caller no longer holds the + * record, whatever its cached view says, so the reason is the one a lost fence already publishes. + */ +function releasedLeaseError(store: DurableCaptureResourceStore): AppError { + return fenceError(store, `${store.displayName} resource fence was released with its run`); +} + +/** + * Whether an error is the refusal a terminal record gives. A caller releasing a resource treats it as + * settled: the record is committed, so the resource is released, whatever the record could replay. + */ +export function isCompletedDurableCaptureError(error: unknown): boolean { + return error instanceof AppError && error.details?.reason === 'resource-already-completed'; +} + +/** + * The refusal a caller gives when its work arrived after the record committed. The reason lives here + * with the rule that produces it, so a finish, a cleanup and a late progress write cannot disagree + * about what a terminal record answers. + */ +export function completedDurableCaptureError(store: DurableCaptureResourceStore): AppError { + return new AppError( + 'COMMAND_FAILED', + `${store.displayName} resource record is already completed`, + { + reason: 'resource-already-completed', + retriable: false, + hint: `Read the completed ${store.displayName.toLowerCase()} record instead of finishing or cleaning it again.`, + }, + ); +} diff --git a/packages/capture-kit/src/durable-capture/finish-recovered.test.ts b/packages/capture-kit/src/durable-capture/finish-recovered.test.ts new file mode 100644 index 0000000000..7d98f9a4a4 --- /dev/null +++ b/packages/capture-kit/src/durable-capture/finish-recovered.test.ts @@ -0,0 +1,83 @@ +import { expect, test, vi } from 'vitest'; +import { createDurableResourceEnvelope } from '../durable-resource-envelope.ts'; +import { + makeDurableCaptureContext, + testCaptureDefinition, + testCaptureStore, + TEST_CAPTURE_KIND, +} from './durable-capture.fixtures.ts'; +import { finishRecoveredDurableCapture } from './finish-recovered.ts'; + +function makeScope() { + return { + signal: new AbortController().signal, + diagnostics: { emit: () => {} }, + progress: { report: () => {} }, + }; +} + +function writeOpenRecord( + resourcePath: string, + context: ReturnType, +) { + testCaptureStore.write( + resourcePath, + createDurableResourceEnvelope({ + resourceKind: TEST_CAPTURE_KIND, + sessionId: context.sessionName, + device: { + id: context.device.id, + family: context.device.platform, + kind: context.device.kind, + }, + owner: context.owner, + fence: context.fence, + lifecycle: 'open', + descriptor: { version: 1, body: { pid: 123 } }, + }), + ); +} + +/** + * Descriptor cleanup is destructive, so terminal admission has to decide before it, not after it ran. + * This stop read the record while it was open and then acquired recovery authority; a live finish + * committed the record during that window. The stop may not remove what that finish owns, so it settles + * without cleaning and keeps its own typed refusal. + */ +test('a recovered stop removes nothing from a record that committed while authority was acquired', async () => { + const context = makeDurableCaptureContext(); + writeOpenRecord(context.resourcePath, context); + const cleanup = vi.fn(async () => ({ status: 'cleaned' }) as const); + + await expect( + finishRecoveredDurableCapture(testCaptureDefinition, { + resourcePath: context.resourcePath, + scope: makeScope(), + acquireControl: async () => { + const record = testCaptureStore.read(context.resourcePath); + if (record.status !== 'decoded') throw new Error('Expected a decodable record'); + testCaptureStore.write(context.resourcePath, { + ...record.envelope, + lifecycle: 'completed', + metadata: { ...(record.envelope.metadata ?? {}), phase: 'completed' }, + }); + return { + reattach: async () => + ({ + status: 'unreattachable', + reason: 'transport-not-reattachable', + message: 'no live control', + }) as const, + cleanup, + [Symbol.asyncDispose]: async () => {}, + }; + }, + }), + ).rejects.toMatchObject({ details: { reason: 'transport-not-reattachable' } }); + + expect(cleanup).not.toHaveBeenCalled(); + expect(testCaptureStore.read(context.resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); diff --git a/packages/capture-kit/src/durable-capture/finish-recovered.ts b/packages/capture-kit/src/durable-capture/finish-recovered.ts index 0f00a619da..1c041b304a 100644 --- a/packages/capture-kit/src/durable-capture/finish-recovered.ts +++ b/packages/capture-kit/src/durable-capture/finish-recovered.ts @@ -11,9 +11,9 @@ import { withDurableCaptureResourceFence } from './fence.ts'; import type { DurableCaptureRecordDefinition } from './definition.ts'; import { capitalizeDurableCaptureLabel, durableCaptureDiagnosticPrefix } from './labels.ts'; import { + cleanupDurableCaptureDescriptor, finishDurableCaptureHandle, requireConfirmedDurableCaptureCleanup, - transitionCleanupOutcome, transitionFinishOutcome, } from './transitions.ts'; @@ -78,6 +78,9 @@ export async function finishRecoveredDurableCapture< store: definition.store, resourcePath: params.resourcePath, expected: envelope.fence, + // The record committed while this recovery was acquiring authority. It holds the export the + // caller is being served, so this attempt writes nothing of its own. + onCompleted: () => undefined, run: async (lease) => { transitionFinishOutcome(definition, lease, { status: 'completed', @@ -99,12 +102,10 @@ export async function finishRecoveredDurableCapture< ); case 'unreattachable': { if (allowsDurableCaptureDescriptorCleanup(authority.reattached.reason)) { - const cleanup = await authority.control.cleanup(envelope); - await withDurableCaptureResourceFence({ - store: definition.store, + const cleanup = await cleanupDurableCaptureDescriptor(definition, { + envelope, resourcePath: params.resourcePath, - expected: envelope.fence, - run: async (lease) => transitionCleanupOutcome(lease, cleanup), + cleanup: () => authority.control.cleanup(envelope), }); requireConfirmedDurableCaptureCleanup(definition, cleanup); } diff --git a/packages/capture-kit/src/durable-capture/index.ts b/packages/capture-kit/src/durable-capture/index.ts index 1d8f84b174..b4c48a597a 100644 --- a/packages/capture-kit/src/durable-capture/index.ts +++ b/packages/capture-kit/src/durable-capture/index.ts @@ -22,4 +22,5 @@ export type { DurableCaptureRecoverySummary, } from './recovery.ts'; export type { DurableCaptureRecoveryControl } from './recovery-authority.ts'; +export { isCompletedDurableCaptureError } from './fence.ts'; export type { DurableCaptureResourceRecord, DurableCaptureResourceStore } from './store.ts'; diff --git a/packages/capture-kit/src/durable-capture/recovery.test.ts b/packages/capture-kit/src/durable-capture/recovery.test.ts index 506a931ae7..2a36cfdc4f 100644 --- a/packages/capture-kit/src/durable-capture/recovery.test.ts +++ b/packages/capture-kit/src/durable-capture/recovery.test.ts @@ -51,3 +51,74 @@ test('generic recovery terminalizes missing native authority through one exact c }, }); }); + +/** + * Descriptor cleanup is destructive, so terminal admission has to decide before it, not after it ran. + * Recovery read the record while it was open, then acquired authority; a finish committed the record + * during that window. Nothing may be removed from a committed record, so this attempt settles without + * touching the descriptor the committing finish owns. + */ +test('generic recovery removes nothing from a record that committed while authority was acquired', async () => { + const context = makeDurableCaptureContext(); + testCaptureStore.write( + context.resourcePath, + createDurableResourceEnvelope({ + resourceKind: TEST_CAPTURE_KIND, + sessionId: context.sessionName, + device: { + id: context.device.id, + family: context.device.platform, + kind: context.device.kind, + }, + owner: context.owner, + fence: context.fence, + lifecycle: 'open', + descriptor: { version: 1, body: { pid: 123 } }, + }), + ); + const cleanup = vi.fn(async () => ({ status: 'cleaned' }) as const); + + await expect( + recoverDurableCaptureResourcesAfterDaemonLock({ + definition: testCaptureDefinition, + sessionsDir: context.sessionsDir, + resolveSessionDir: context.resolveSessionDir, + scope: { + signal: new AbortController().signal, + diagnostics: { emit: () => {} }, + progress: { report: () => {} }, + }, + acquireControl: async () => { + // Another caller commits while this recovery holds authority and has not cleaned yet. + commitCommittedRecord(context.resourcePath); + return { + reattach: async () => + ({ + status: 'unreattachable', + reason: 'transport-not-reattachable', + message: 'no live control', + }) as const, + cleanup, + [Symbol.asyncDispose]: async () => {}, + }; + }, + }), + ).resolves.toEqual({ scanned: 1, recovered: 1, retained: 0 }); + + expect(cleanup).not.toHaveBeenCalled(); + expect(testCaptureStore.read(context.resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); + +/** Commits the record a recovery is working on, through the same fence that record names. */ +function commitCommittedRecord(resourcePath: string): void { + const record = testCaptureStore.read(resourcePath); + if (record.status !== 'decoded') throw new Error('Expected a decodable test capture record'); + testCaptureStore.write(resourcePath, { + ...record.envelope, + lifecycle: 'completed', + metadata: { ...(record.envelope.metadata ?? {}), phase: 'completed' }, + }); +} diff --git a/packages/capture-kit/src/durable-capture/recovery.ts b/packages/capture-kit/src/durable-capture/recovery.ts index 4f09648e0a..821841fced 100644 --- a/packages/capture-kit/src/durable-capture/recovery.ts +++ b/packages/capture-kit/src/durable-capture/recovery.ts @@ -13,10 +13,14 @@ import { type DurableCaptureRecoveryAuthority, type DurableCaptureRecoveryControl, } from './recovery-authority.ts'; -import { withDurableCaptureResourceFence, type DurableCaptureResourceFenceLease } from './fence.ts'; +import { withDurableCaptureResourceFence } from './fence.ts'; import type { DurableCaptureRecordDefinition } from './definition.ts'; import { durableCaptureDiagnosticPrefix } from './labels.ts'; -import { transitionCleanupOutcome } from './transitions.ts'; +import { + cleanupDurableCaptureDescriptor, + markCompletingDurableCapture, + transitionCleanupOutcome, +} from './transitions.ts'; const DEFAULT_RECOVERY_DEADLINE_MS = 5_000; @@ -162,8 +166,12 @@ async function settleRecoveryAuthority await handle.forceCleanup(), run: async (lease) => { - markCompleting(lease); + markCompletingDurableCapture(lease); const outcome = await handle.forceCleanup(); transitionCleanupOutcome(lease, outcome); return outcome; @@ -189,12 +197,10 @@ async function settleRecoveryAuthority transitionCleanupOutcome(lease, cleanup), + cleanup: () => authority.control.cleanup(envelope), }); return isConfirmedCleanup(cleanup); } @@ -214,6 +220,9 @@ async function transitionTerminal undefined, run: async (lease) => { lease.transition('completed', { metadata: { ...(lease.envelope.metadata ?? {}), phase: 'completed', ...metadata }, @@ -222,12 +231,6 @@ async function transitionTerminal(lease: DurableCaptureResourceFenceLease): void { - lease.transition('open', { - metadata: { ...(lease.envelope.metadata ?? {}), phase: 'completing' }, - }); -} - export function allowsDurableCaptureDescriptorCleanup( reason: ResourceUnreattachableReason, ): boolean { diff --git a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts new file mode 100644 index 0000000000..dc14f427b7 --- /dev/null +++ b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts @@ -0,0 +1,372 @@ +import { expect, test, vi } from 'vitest'; +import fs from 'node:fs'; +import { PendingTransferGuard } from '@agent-device/contracts/async-lifecycle'; +import { + flushDiagnosticsToSessionFile, + withDiagnosticsScope, +} from '@agent-device/host-kit/diagnostics'; +import { createAppLogLiveHandle } from '../app-log-live-handle.ts'; +import type { CleanupOutcome } from '@agent-device/contracts/durable-resource'; +import { adoptStartedDurableCapture } from './adoption.ts'; +import { + createTestCaptureDefinition, + makeDurableCaptureContext, + makeDurableCaptureStartResult, + testCaptureDefinition, + testCaptureStore, + type TestCaptureHandle, +} from './durable-capture.fixtures.ts'; +import { + forceCleanupLiveDurableCapture, + finishLiveDurableCapture, + finishDurableCaptureHandle, +} from './transitions.ts'; + +test.each(['replay', 'refusal'] as const)( + 'a terminal %s cleans the captured handle another handle did not release', + async (outcome) => { + const definition = { + ...testCaptureDefinition, + completionFromMetadata: + outcome === 'replay' ? testCaptureDefinition.completionFromMetadata : undefined, + }; + const { context, captured, completion, record } = + await captureWhoseRecordAnotherHandleCommitted(definition); + + const finish = finishLiveDurableCapture( + definition, + { binding: context.binding, intent: 'disposal' }, + context.resourcePath, + ); + if (outcome === 'replay') await expect(finish).resolves.toEqual(completion); + else + await expect(finish).rejects.toMatchObject({ + details: { reason: 'resource-already-completed' }, + }); + + expect(captured.forceCleanup).toHaveBeenCalledOnce(); + expect(captured.finish).not.toHaveBeenCalled(); + expect(context.binding.read()).toBeUndefined(); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); + }, +); + +test.each(['replay', 'refusal'] as const)( + 'capture intent retains the %s binding after rejected cleanup for a later successful teardown', + async (outcome) => { + const definition = { + ...testCaptureDefinition, + completionFromMetadata: + outcome === 'replay' ? testCaptureDefinition.completionFromMetadata : undefined, + }; + const { context, captured, completion, record } = + await captureWhoseRecordAnotherHandleCommitted(definition); + const cleanup = vi + .spyOn(captured.handle, 'forceCleanup') + .mockRejectedValueOnce(new Error('Captured handle cleanup failed')); + const logPath = `${context.resourcePath}.diagnostics.ndjson`; + const finish = (intent: 'capture' | 'disposal' = 'capture') => + finishLiveDurableCapture( + definition, + { binding: context.binding, intent }, + context.resourcePath, + ); + await withDiagnosticsScope({ command: 'record', logPath }, async () => { + const [first] = await Promise.allSettled([finish()]); + expect(context.binding.read()?.handle).toBe(captured.handle); + if (outcome === 'replay') expect(first).toEqual({ status: 'fulfilled', value: completion }); + else + expect(first).toMatchObject({ + status: 'rejected', + reason: { details: { reason: 'cleanup-unconfirmed', cleanupStatus: 'cleanup-pending' } }, + }); + flushDiagnosticsToSessionFile({ force: true }); + }); + const events = fs + .readFileSync(logPath, 'utf8') + .trim() + .split('\n') + .map((line) => JSON.parse(line)); + expect(events).toContainEqual( + expect.objectContaining({ + level: 'warn', + phase: 'test_capture_terminal_cleanup_pending', + data: expect.objectContaining({ + cleanup: { + status: 'cleanup-pending', + reason: 'cleanup-unconfirmed', + message: 'Captured handle cleanup failed', + }, + }), + }), + ); + if (outcome === 'replay') await expect(finish('disposal')).resolves.toEqual(completion); + else + await expect(finish('disposal')).rejects.toMatchObject({ + details: { reason: 'resource-already-completed' }, + }); + expect(cleanup).toHaveBeenCalledTimes(2); + expect(captured.finish).not.toHaveBeenCalled(); + expect(context.binding.read()).toBeUndefined(); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); + }, +); + +test.each(['rejected', 'unconfirmed'] as const)( + 'disposal intent reports %s terminal cleanup even when the committed result is replayable', + async (failure) => { + const cleanup = vi.fn<() => Promise>().mockResolvedValue({ status: 'cleaned' }); + if (failure === 'rejected') cleanup.mockRejectedValueOnce(new Error('Transport unavailable')); + else cleanup.mockResolvedValueOnce({ status: 'cleanup-pending', reason: 'transport-failed' }); + const handle = createAppLogLiveHandle({ + inspect: () => ({ backend: 'android', state: 'active', startedAt: 1 }), + finish: async () => { + throw new Error('Terminal disposal must not re-finish'); + }, + forceCleanup: cleanup, + }); + const { context, completion, record } = await captureWhoseRecordAnotherHandleCommitted( + testCaptureDefinition, + handle, + ); + const finish = () => + finishLiveDurableCapture( + testCaptureDefinition, + { binding: context.binding, intent: 'disposal' }, + context.resourcePath, + ); + await expect(finish()).rejects.toMatchObject({ + code: 'COMMAND_FAILED', + details: { + reason: failure === 'rejected' ? 'cleanup-unconfirmed' : 'transport-failed', + cleanupStatus: 'cleanup-pending', + }, + }); + expect(context.binding.read()?.handle).toBe(handle); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); + await expect(finish()).resolves.toEqual(completion); + expect(cleanup).toHaveBeenCalledTimes(2); + expect(context.binding.read()).toBeUndefined(); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); + }, +); + +test.each(['rejected', 'unconfirmed'] as const)( + 'a retained real app-log handle retries its %s disposal', + async (failure) => { + const cleanup = vi.fn<() => Promise>().mockResolvedValue({ status: 'cleaned' }); + if (failure === 'rejected') cleanup.mockRejectedValueOnce(new Error('Transport unavailable')); + else cleanup.mockResolvedValueOnce({ status: 'cleanup-pending', reason: 'transport-failed' }); + const handle = createAppLogLiveHandle({ + inspect: () => ({ backend: 'android', state: 'active', startedAt: 1 }), + finish: async () => { + throw new Error('Terminal replay must not re-finish'); + }, + forceCleanup: cleanup, + }); + const { context, completion, record } = await captureWhoseRecordAnotherHandleCommitted( + testCaptureDefinition, + handle, + ); + const finish = () => + finishLiveDurableCapture( + testCaptureDefinition, + { binding: context.binding, intent: 'capture' }, + context.resourcePath, + ); + await expect(finish()).resolves.toEqual(completion); + expect(context.binding.read()?.handle).toBe(handle); + await expect(finish()).resolves.toEqual(completion); + expect(cleanup).toHaveBeenCalledTimes(2); + expect(context.binding.read()).toBeUndefined(); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); + }, +); + +async function captureWhoseRecordAnotherHandleCommitted( + definition: typeof testCaptureDefinition, + handle?: TestCaptureHandle, +) { + const context = makeDurableCaptureContext(); + const start = makeDurableCaptureStartResult(context); + const captured = handle + ? { ...start, handle, pendingHandle: new PendingTransferGuard(handle) } + : start; + await adoptStartedDurableCapture( + definition, + { ...context, ...captured, throwIfCanceled: () => {} }, + context.resourcePath, + ); + const committing = makeDurableCaptureStartResult(context); + const completion = await finishDurableCaptureHandle(definition, { + handle: committing.handle, + fence: context.fence, + resourcePath: context.resourcePath, + intent: 'capture', + }); + return { context, captured, completion, record: testCaptureStore.read(context.resourcePath) }; +} + +/** + * A cleanup that reaches the fence only after a finish committed the same record. The cleanup had + * read the session's active resource before that finish, so it holds the fence the record still + * names; ownership alone must not let it write the record back open. It still signals the handle it + * captured, because that handle is a native resource of its own that nobody else has disposed. + */ +test('a cleanup queued behind a committed finish leaves the record completed and settles its handle', async () => { + const context = makeDurableCaptureContext(); + const start = makeDurableCaptureStartResult(context); + await adoptStartedDurableCapture( + testCaptureDefinition, + { ...context, ...start, throwIfCanceled: () => {} }, + context.resourcePath, + ); + // The queued cleanup's own view of the session, captured while the record was still open. + const queued = context.binding.read(); + if (!queued) throw new Error('Expected adopted test capture session'); + + await finishLiveDurableCapture( + testCaptureDefinition, + { binding: context.binding, intent: 'capture' }, + context.resourcePath, + ); + expect(testCaptureStore.read(context.resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); + + // The queued cleanup now reaches the same slot and the same fence its record still names. + context.binding.adopt(queued); + await forceCleanupLiveDurableCapture(testCaptureDefinition, { + binding: context.binding, + resourcePath: context.resourcePath, + }); + + expect(start.forceCleanup).toHaveBeenCalledOnce(); + expect(start.finish).toHaveBeenCalledOnce(); + expect(testCaptureStore.read(context.resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); + expect(context.binding.read()).toBeUndefined(); +}); + +/** + * Two finishes captured the same live resource. The first commits its export; the second reaches the + * fence after that commit. It previously succeeded by replaying the handle's memoized finish, so a + * refusal here is a new user-visible failure on a path that already has the export to serve — the + * teardown path reports it as an unreleased resource. It answers the committed completion and drives + * the native finish no second time. + */ +test('a finish queued behind a committed finish replays the committed export without re-finishing', async () => { + const context = makeDurableCaptureContext(); + const start = makeDurableCaptureStartResult(context); + await adoptStartedDurableCapture( + testCaptureDefinition, + { ...context, ...start, throwIfCanceled: () => {} }, + context.resourcePath, + ); + const queued = context.binding.read(); + if (!queued) throw new Error('Expected adopted test capture session'); + + const first = await finishLiveDurableCapture( + testCaptureDefinition, + { binding: context.binding, intent: 'capture' }, + context.resourcePath, + ); + + context.binding.adopt(queued); + await expect( + finishLiveDurableCapture( + testCaptureDefinition, + { binding: context.binding, intent: 'disposal' }, + context.resourcePath, + ), + ).resolves.toEqual(first); + expect(start.finish).toHaveBeenCalledOnce(); + expect(context.binding.read()).toBeUndefined(); +}); + +/** + * A failed finish that confirmed its compensating cleanup committed a record with no export in it. + * Nothing is serveable from it, so the caller keeps its own error rather than a made-up completion. + */ +test('a finish queued behind a commit that produced no export keeps its refusal', async () => { + const context = makeDurableCaptureContext(); + const start = makeDurableCaptureStartResult(context); + await adoptStartedDurableCapture( + testCaptureDefinition, + { ...context, ...start, throwIfCanceled: () => {} }, + context.resourcePath, + ); + const queued = context.binding.read(); + if (!queued) throw new Error('Expected adopted test capture session'); + // A record that committed without ever storing this kind's completion. + testCaptureStore.write(context.resourcePath, { + ...( + testCaptureStore.read(context.resourcePath) as Extract< + ReturnType, + { status: 'decoded' } + > + ).envelope, + lifecycle: 'completed', + metadata: { phase: 'completed', cleanupStatus: 'cleaned' }, + }); + + await expect( + finishLiveDurableCapture( + testCaptureDefinition, + { binding: context.binding, intent: 'disposal' }, + context.resourcePath, + ), + ).rejects.toMatchObject({ details: { reason: 'resource-already-completed' } }); + expect(start.finish).not.toHaveBeenCalled(); +}); + +/** + * A failed export under a preserving kind has to be drivable again: the retry reads the open record + * and its journaled facts. Terminal admission must not treat an in-flight attempt as terminal. + */ +test('a failed export under a preserving kind stays retryable through the same fence', async () => { + const context = makeDurableCaptureContext(); + const definition = createTestCaptureDefinition(testCaptureStore, 'preserve-retry-material'); + const start = makeDurableCaptureStartResult(context, { + finishError: new Error('collect failed'), + }); + await adoptStartedDurableCapture( + definition, + { ...context, ...start, throwIfCanceled: () => {} }, + context.resourcePath, + ); + + await expect( + finishLiveDurableCapture( + definition, + { binding: context.binding, intent: 'capture' }, + context.resourcePath, + ), + ).rejects.toThrow('collect failed'); + expect(testCaptureStore.read(context.resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'open', metadata: { phase: 'completing' } }, + }); + expect(context.binding.read()).toBeDefined(); + + // The same live handle drives the stop again, and this time its export commits. + start.finish.mockImplementationOnce(async () => ({ + status: 'completed', + result: { outputPath: '/tmp/capture', completedAt: 3 }, + })); + await expect( + finishLiveDurableCapture( + definition, + { binding: context.binding, intent: 'capture' }, + context.resourcePath, + ), + ).resolves.toMatchObject({ outputPath: '/tmp/capture', completedAt: 3 }); + expect(start.forceCleanup).not.toHaveBeenCalled(); + expect(testCaptureStore.read(context.resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); diff --git a/packages/capture-kit/src/durable-capture/transitions.ts b/packages/capture-kit/src/durable-capture/transitions.ts index 3536afe481..6c82c9cef1 100644 --- a/packages/capture-kit/src/durable-capture/transitions.ts +++ b/packages/capture-kit/src/durable-capture/transitions.ts @@ -6,9 +6,15 @@ import { isConfirmedCleanup, } from '@agent-device/contracts/durable-resource'; import type { JsonObject } from '@agent-device/contracts/client'; +import type { ResourceOwnershipFence } from '@agent-device/contracts/platform-runtime'; +import type { DurableResourceEnvelope } from '@agent-device/contracts/durable-resource-envelope'; import { AppError, errorMessage } from '@agent-device/kernel/errors'; import { emitDiagnostic } from '@agent-device/host-kit/diagnostics'; -import { withDurableCaptureResourceFence, type DurableCaptureResourceFenceLease } from './fence.ts'; +import { + completedDurableCaptureError, + withDurableCaptureResourceFence, + type DurableCaptureOpenResourceFenceLease, +} from './fence.ts'; import type { DurableCaptureFinishIntent, DurableCaptureRecordDefinition, @@ -30,18 +36,26 @@ export async function finishLiveDurableCapture< ): Promise { const active = params.binding.read(); if (!active) throw new AppError('INVALID_ARGS', definition.messages.noActive); + let cleanupPending = false; try { const result = await finishDurableCaptureHandle(definition, { handle: active.handle, fence: active.envelope.fence, resourcePath, intent: params.intent, + onCleanupPending: () => { + cleanupPending = true; + }, }); - params.binding.clear(active); + if (!cleanupPending) params.binding.clear(active); return result; } catch (error) { const record = definition.store.read(resourcePath); - if (record.status === 'decoded' && record.envelope.lifecycle === 'completed') { + if ( + !cleanupPending && + record.status === 'decoded' && + record.envelope.lifecycle === 'completed' + ) { params.binding.clear(active); } throw error; @@ -56,17 +70,47 @@ export async function finishDurableCaptureHandle< definition: DurableCaptureRecordDefinition, params: { handle: H; - fence: DurableCaptureResourceFenceLease['envelope']['fence']; + fence: ResourceOwnershipFence; resourcePath: string; intent: DurableCaptureFinishIntent; + onCleanupPending?(outcome: Extract): void; }, ): Promise { return await withDurableCaptureResourceFence({ store: definition.store, resourcePath: params.resourcePath, expected: params.fence, + // The record this fence found committed already holds an outcome. A finish arriving after that is + // not a failure to report: it is a caller that needs what the first finish produced, and the record + // is the only place it can come from. A queued teardown finish that had to be refused this way + // reported the session's export as an unreleased resource. + onCompleted: async ({ envelope }) => { + let cleanup: CleanupOutcome; + try { + cleanup = await params.handle.forceCleanup(); + } catch (error) { + cleanup = { + status: 'cleanup-pending', + reason: 'cleanup-unconfirmed', + message: errorMessage(error), + }; + } + if (!isConfirmedCleanup(cleanup)) { + params.onCleanupPending?.(cleanup); + emitDiagnostic({ + level: 'warn', + phase: `${durableCaptureDiagnosticPrefix(definition.resourceKind)}_terminal_cleanup_pending`, + data: { resourcePath: params.resourcePath, cleanup }, + }); + if (params.intent === 'disposal') throw cleanupPendingError(definition, cleanup); + } + const replayed = await definition.completionFromMetadata?.(envelope.metadata); + if (replayed !== undefined) return replayed; + if (!isConfirmedCleanup(cleanup)) throw cleanupPendingError(definition, cleanup); + throw completedDurableCaptureError(definition.store); + }, run: async (lease) => { - markCompleting(lease); + markCompletingDurableCapture(lease); const progress = createStopProgress(lease); let finishOutcome: FinishOutcome; try { @@ -89,7 +133,7 @@ export async function finishDurableCaptureHandle< async function respondToFailedFinish, C>( definition: DurableCaptureRecordDefinition, - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, params: { handle: H; resourcePath: string }, finishError: unknown, intent: DurableCaptureFinishIntent, @@ -131,7 +175,7 @@ async function respondToFailedFinish( - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, cleanupError: unknown, ): unknown | undefined { try { @@ -199,8 +243,13 @@ export async function forceCleanupLiveDurableCapture< store: definition.store, resourcePath: params.resourcePath, expected: active.envelope.fence, + // The record is settled, but the handle this caller captured is still a native resource of its + // own: a commit through another entry point (recovery, or a finish this caller never saw) leaves + // it live here. Signal it and report what it answers, without letting it touch the record — a + // kind whose cleanup is already done after its own commit answers confirmed. + onCompleted: async () => await active.handle.forceCleanup(), run: async (lease) => { - markCompleting(lease); + markCompletingDurableCapture(lease); const result = await active.handle.forceCleanup(); transitionCleanupOutcome(lease, result); return result; @@ -210,8 +259,39 @@ export async function forceCleanupLiveDurableCapture< params.binding.clear(active); } +/** + * Removes what a persisted record points at when its native resource could not be reattached, and + * journals that removal. + * + * The cleanup is destructive, so it runs inside the fence's `run`, not before it: acquiring recovery + * authority can take long enough for another caller to commit the record, and removal that arrives + * after that commit has nothing left to remove. The terminal branch therefore answers `already-missing` + * — what a committed record states about its own descriptor, since the commit that produced it confirmed + * its cleanup — and removes nothing. + */ +export async function cleanupDurableCaptureDescriptor( + definition: DurableCaptureRecordDefinition, + params: { + envelope: DurableResourceEnvelope; + resourcePath: string; + cleanup(): Promise; + }, +): Promise { + return await withDurableCaptureResourceFence({ + store: definition.store, + resourcePath: params.resourcePath, + expected: params.envelope.fence, + onCompleted: () => ({ status: 'already-missing' as const }), + run: async (lease) => { + const outcome = await params.cleanup(); + transitionCleanupOutcome(lease, outcome); + return outcome; + }, + }); +} + export function transitionCleanupOutcome( - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, outcome: CleanupOutcome, ): void { lease.transition(isConfirmedCleanup(outcome) ? 'completed' : 'open', { @@ -246,6 +326,7 @@ function cleanupPendingError( `${capitalizeDurableCaptureLabel(definition.displayName)} cleanup could not be confirmed`, { reason: outcome.reason, + cleanupStatus: outcome.status, retriable: outcome.reason !== 'ownership-fence-lost', hint: definition.messages.cleanupPendingHint, }, @@ -254,7 +335,7 @@ function cleanupPendingError( export function transitionFinishOutcome( definition: DurableCaptureRecordDefinition, - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, outcome: FinishOutcome, ): void { if (outcome.status === 'completed') { @@ -277,7 +358,7 @@ export function transitionFinishOutcome( * record and the next attempt reads them instead of redoing durable work. */ function createStopProgress( - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, ): DurableCaptureProgress { return Object.freeze({ learned: lease.envelope.metadata, @@ -287,7 +368,14 @@ function createStopProgress( }); } -function markCompleting(lease: DurableCaptureResourceFenceLease): void { +/** + * The fact that an attempt is in flight, journaled into a record this fence found open. Recovery + * reads it to tell an abandoned attempt from one that never started, so it belongs to every entry + * that drives a record toward completion, not to one of them. + */ +export function markCompletingDurableCapture( + lease: DurableCaptureOpenResourceFenceLease, +): void { lease.transition('open', { metadata: { ...(lease.envelope.metadata ?? {}), phase: 'completing' }, }); diff --git a/packages/capture-kit/src/index.ts b/packages/capture-kit/src/index.ts index 5c83dd4f4e..afc77befd0 100644 --- a/packages/capture-kit/src/index.ts +++ b/packages/capture-kit/src/index.ts @@ -10,6 +10,7 @@ export { export { createAppLogLiveHandle, createAppLogLiveHandleFromFinish } from './app-log-live-handle.ts'; export { createHostAudioProbeCaptureOperations } from './audio-probe-runtime.ts'; export { decodeDurableDescriptor } from './durable-descriptor-codec.ts'; +export { createRetryableCleanup } from '@agent-device/contracts/durable-resource'; export { createScreenRecordingLiveHandle } from './screen-recording-live-handle.ts'; export { createScreenRecordingCompletion } from './screen-recording-completion.ts'; export { assertScreenRecordingOptionsSupported } from './screen-recording-options.ts'; diff --git a/packages/capture-kit/src/provider-app-log-poller.test.ts b/packages/capture-kit/src/provider-app-log-poller.test.ts index 55a86e33f7..c4c3f8e29c 100644 --- a/packages/capture-kit/src/provider-app-log-poller.test.ts +++ b/packages/capture-kit/src/provider-app-log-poller.test.ts @@ -122,18 +122,18 @@ describe('provider app-log poller', () => { test('still disposes the output when reader cleanup rejects', async () => { const sleeps = deferredSleeps(); let outputDisposed = false; + const onOutputDispose = vi.fn(() => { + outputDisposed = true; + }); + const dispose = vi.fn(async () => {}).mockRejectedValueOnce(new Error('reader cleanup failed')); const reader = providerReader({ - [Symbol.asyncDispose]: async () => { - throw new Error('reader cleanup failed'); - }, + [Symbol.asyncDispose]: dispose, }); const handle = await startProviderAppLogPoller( pollerOptions(reader, { writes: [], sleeps, - onOutputDispose: () => { - outputDisposed = true; - }, + onOutputDispose, }), ); await vi.waitFor(() => expect(sleeps.hasPending(1_000)).toBe(true)); @@ -146,6 +146,10 @@ describe('provider app-log poller', () => { }); expect(outputDisposed).toBe(true); expect(handle.inspect().state).toBe('failed'); + await handle[Symbol.asyncDispose](); + await handle[Symbol.asyncDispose](); + expect(dispose).toHaveBeenCalledTimes(2); + expect(onOutputDispose).toHaveBeenCalledOnce(); }); test('abandons a read that ignores its signal once the budget elapses', async () => { diff --git a/packages/capture-kit/src/provider-app-log-poller.ts b/packages/capture-kit/src/provider-app-log-poller.ts index 66432492c2..b5d21e3c89 100644 --- a/packages/capture-kit/src/provider-app-log-poller.ts +++ b/packages/capture-kit/src/provider-app-log-poller.ts @@ -89,6 +89,7 @@ function createPollerHandle( await host.clock.sleep(POLL_INTERVAL_MS); } })(); + const pendingDisposals = new Set([options.reader, output]); let finishPromise: | Promise> | undefined; @@ -96,8 +97,9 @@ function createPollerHandle( (finishPromise ??= (async () => { stopped = true; await polling; - const failures = await disposeAll([options.reader, output]); + const failures = await disposeAll(pendingDisposals); if (failures.length > 0) { + finishPromise = undefined; state = 'failed'; return { status: 'cleanup-pending', @@ -171,9 +173,12 @@ function remoteOnlyTail(tail: string): string { .join('\n'); } -async function disposeAll(resources: readonly AsyncDisposable[]): Promise { +async function disposeAll(resources: Set): Promise { const results = await Promise.allSettled( - resources.map(async (resource) => await resource[Symbol.asyncDispose]()), + [...resources].map(async (resource) => { + await resource[Symbol.asyncDispose](); + resources.delete(resource); + }), ); return results.flatMap((result) => (result.status === 'rejected' ? [result.reason] : [])); } diff --git a/packages/capture-kit/src/screen-recording-live-handle.test.ts b/packages/capture-kit/src/screen-recording-live-handle.test.ts index 32550db74a..239bf00ccd 100644 --- a/packages/capture-kit/src/screen-recording-live-handle.test.ts +++ b/packages/capture-kit/src/screen-recording-live-handle.test.ts @@ -1,6 +1,32 @@ +import { AppError } from '@agent-device/kernel/errors'; import assert from 'node:assert/strict'; -import { test, vi } from 'vitest'; +import { expect, test, vi } from 'vitest'; import { createScreenRecordingLiveHandle } from './screen-recording-live-handle.ts'; +import type { CleanupOutcome } from '@agent-device/contracts/durable-resource'; + +test.each(['rejected', 'unconfirmed'] as const)( + 'disposal retries %s cleanup and stays inert after confirmation', + async (failure) => { + const cleanup = vi.fn<() => Promise>().mockResolvedValue({ status: 'cleaned' }); + if (failure === 'rejected') + cleanup.mockRejectedValueOnce( + new AppError('COMMAND_FAILED', 'Transport unavailable', { reason: 'transport-failed' }), + ); + else cleanup.mockResolvedValueOnce({ status: 'cleanup-pending', reason: 'transport-failed' }); + const finish = vi.fn(async () => completed()); + const handle = createScreenRecordingLiveHandle(snapshot(), { finish, forceCleanup: cleanup }); + await expect(Promise.resolve(handle[Symbol.asyncDispose]())).rejects.toMatchObject({ + code: 'COMMAND_FAILED', + details: { reason: 'transport-failed' }, + }); + handle.appendGestureEvents([{ kind: 'tap', tMs: 3, x: 4, y: 5 }]); + assert.deepEqual(handle.inspect().gestureEvents, []); + await Promise.all([handle[Symbol.asyncDispose](), handle[Symbol.asyncDispose]()]); + await handle[Symbol.asyncDispose](); + assert.equal(cleanup.mock.calls.length, 2); + assert.equal(finish.mock.calls.length, 0); + }, +); test('keeps mutable gesture evidence on the live handle and settles cleanup once', async () => { let cleanupCalls = 0; diff --git a/packages/capture-kit/src/screen-recording-live-handle.ts b/packages/capture-kit/src/screen-recording-live-handle.ts index 5852d82975..14ce27af82 100644 --- a/packages/capture-kit/src/screen-recording-live-handle.ts +++ b/packages/capture-kit/src/screen-recording-live-handle.ts @@ -4,6 +4,7 @@ import { type DurableCaptureProgress, type FinishOutcome, isConfirmedCleanup, + createRetryableCleanup, } from '@agent-device/contracts/durable-resource'; import type { RecordingGestureEvent, @@ -34,23 +35,21 @@ export function createScreenRecordingLiveHandle( ): ScreenRecordingLiveHandle { let snapshot = freezeSnapshot(initial); let finish: Promise> | undefined; - let finishRequested = false; - let cleanup: Promise | undefined; - let disposal: Promise | undefined; + let terminalRequested = false; // Once a finish or a cleanup has been asked for, the recording no longer accepts telemetry. - const terminal = (): boolean => finishRequested || cleanup !== undefined; + const terminal = (): boolean => terminalRequested; const finishRecording = (progress?: DurableCaptureProgress) => { // Only a finished recording stays memoized. A finish the host refused has to be re-driven // by the next `record stop`, which is how a recording recovers without a daemon restart. - finishRequested = true; + terminalRequested = true; finish ??= implementation.finish(snapshot, progress).catch((error: unknown) => { finish = undefined; throw error; }); return finish; }; - const forceCleanup = () => - (cleanup ??= finish + const clean = () => + finish ? finish.then( async (outcome) => outcome.status === 'completed' @@ -58,7 +57,7 @@ export function createScreenRecordingLiveHandle( : await implementation.forceCleanup(snapshot), async () => await implementation.forceCleanup(snapshot), ) - : implementation.forceCleanup(snapshot)); + : implementation.forceCleanup(snapshot); return Object.freeze({ inspect: () => snapshot, appendGestureEvents: (events: readonly RecordingGestureEvent[]) => { @@ -84,11 +83,10 @@ export function createScreenRecordingLiveHandle( snapshot = freezeSnapshot({ ...snapshot, invalidatedReason }); }, finish: finishRecording, - forceCleanup, - [Symbol.asyncDispose]: async () => { - disposal ??= forceCleanup().then(assertConfirmedCleanup); - await disposal; - }, + ...createRetryableCleanup(() => { + terminalRequested = true; + return clean(); + }, assertConfirmedCleanup), }); } diff --git a/packages/contracts/src/durable-resource.ts b/packages/contracts/src/durable-resource.ts index 45562422f1..4379409c5c 100644 --- a/packages/contracts/src/durable-resource.ts +++ b/packages/contracts/src/durable-resource.ts @@ -66,3 +66,26 @@ export function isConfirmedCleanup( ): outcome is Extract { return outcome.status === 'cleaned' || outcome.status === 'already-missing'; } + +/** Shares in-flight cleanup and retains only confirmed release across cleanup and disposal. */ +export function createRetryableCleanup( + clean: () => Promise, + assertConfirmed: (outcome: CleanupOutcome) => void, +) { + let cleanup: Promise | undefined; + const forceCleanup = (): Promise => + (cleanup ??= clean().then( + (outcome) => { + if (!isConfirmedCleanup(outcome)) cleanup = undefined; + return outcome; + }, + (error) => { + cleanup = undefined; + throw error; + }, + )); + return { + forceCleanup, + [Symbol.asyncDispose]: async () => assertConfirmed(await forceCleanup()), + }; +} diff --git a/packages/daemon/src/sessions/__tests__/session-teardown-resource-order.test.ts b/packages/daemon/src/sessions/__tests__/session-teardown-resource-order.test.ts index 36463075bc..c8e9aa5944 100644 --- a/packages/daemon/src/sessions/__tests__/session-teardown-resource-order.test.ts +++ b/packages/daemon/src/sessions/__tests__/session-teardown-resource-order.test.ts @@ -3,6 +3,7 @@ import { bindTestCommandHost, createTestHostEnvironment } from '@agent-device/te import { createDurableResourceEnvelope } from '@agent-device/capture-kit'; import { localRuntimeOwner } from '@agent-device/contracts/platform-runtime'; import { deviceIdentity } from '@agent-device/kernel/device'; +import { AppError } from '@agent-device/kernel/errors'; import { sessionCloseShutdownFixture, type SessionState, @@ -12,11 +13,23 @@ import { daemonPlatformServicesFixture } from '../../__tests__/support/platform- import { fakeCreateCommandSurface } from '../../__tests__/support/command-surface-fake.ts'; import { lifecycleOperationsSpy } from '../../__tests__/support/lifecycle-operations-spy.ts'; -const stopEvents = vi.hoisted(() => ({ order: [] as string[], failing: new Set() })); +const stopEvents = vi.hoisted(() => ({ + order: [] as string[], + failing: new Set(), + terminal: new Set(), +})); function recordStop(step: string): void { stopEvents.order.push(step); if (stopEvents.failing.has(step)) throw new Error(`${step} stop failed`); + if (stopEvents.terminal.has(step)) { + // The refusal the durable mechanics publish for a record that had already committed, built here + // from the typed reason so this test does not depend on a test-only production export. + throw new AppError('COMMAND_FAILED', 'Audio probe resource record is already completed', { + reason: 'resource-already-completed', + retriable: false, + }); + } } vi.mock('@agent-device/capture-kit/screen-recording-session-resource', async (importOriginal) => ({ @@ -59,6 +72,7 @@ beforeEach(() => { resetSessionCloseShutdownMocks(); stopEvents.order.length = 0; stopEvents.failing.clear(); + stopEvents.terminal.clear(); }); function sessionWithEveryResource( @@ -208,3 +222,21 @@ test('daemon teardown omits the app-log stop once the app-log fence has settled' 'platform_snapshot_helper', ]); }); + +/** + * A stop that reached a resource whose durable record had already committed is told the record is + * terminal. The resource is released by that commit, so teardown must not report it among the + * resources it failed to release, and must keep stopping what comes after it. + */ +test.each(CALLERS)( + '$name settles a resource whose record already committed instead of reporting it unreleased', + async ({ name, run }) => { + stopEvents.terminal.add('audio_probe'); + + const result = await run({ sessionName: 'stop-terminal-session' }); + if (name === 'ordinary close') expect(result).toMatchObject({ ok: true }); + else expect(result).toBeUndefined(); + + expect(stopEvents.order).toEqual(COMMON_RESOURCE_ORDER); + }, +); diff --git a/packages/daemon/src/sessions/session-teardown.ts b/packages/daemon/src/sessions/session-teardown.ts index 4782e9f1ea..254ccff88e 100644 --- a/packages/daemon/src/sessions/session-teardown.ts +++ b/packages/daemon/src/sessions/session-teardown.ts @@ -1,4 +1,5 @@ import { AppError } from '@agent-device/kernel/errors'; +import { isCompletedDurableCaptureError } from '@agent-device/capture-kit/durable-capture'; import { emitDiagnostic } from '@agent-device/host-kit/diagnostics'; import { cleanupRetainedMaterializedPathsForSession } from './materialized-path-registry.ts'; import type { SessionRef, SessionState } from '../core/session-state.ts'; @@ -122,7 +123,9 @@ export function sessionResourceStopSteps(params: { }): SessionCleanupStep[] { const { ref, sessionStore, platformCleanup } = params; const session = sessionStore.resolveCurrent(ref) ?? ref.session; - return [ + // The session's durable-capture stops. Each one can be refused by a record that committed while the + // step was queued, and that refusal settles: the stopping belongs to the commit, not to this step. + const durableCaptureStops: SessionCleanupStep[] = [ // Finalize any still-active recording BEFORE the Apple runner is stopped: the runner supplies // gesture-telemetry for overlay finalization, and signalling the recorder first prevents a // leaked `simctl recordVideo` child (and its 0-byte, slot-holding mp4) when a session is torn @@ -133,6 +136,10 @@ export function sessionResourceStopSteps(params: { : []), { step: 'audio_probe', run: () => finishSessionAudioProbe({ ref, sessionStore }) }, { step: 'perf_capture', run: () => stopSessionPerfCapture({ ref, sessionStore }) }, + ]; + return [ + ...durableCaptureStops.map(settleCommittedRecordStop), + // The snapshot helper owns no durable record, so its failures stay its own. { step: 'platform_snapshot_helper', run: () => stopSessionSnapshotHelper(session, platformCleanup), @@ -140,6 +147,24 @@ export function sessionResourceStopSteps(params: { ]; } +/** + * Makes one cleanup step settle when the durable record it stopped against answers that the record is + * already completed. That refusal means the resource was released by the stop that committed the + * record; reporting it as unreleased would send an operator after a resource that is already gone. + */ +function settleCommittedRecordStop(step: SessionCleanupStep): SessionCleanupStep { + return { + ...step, + run: async () => { + try { + await step.run(); + } catch (error) { + if (!isCompletedDurableCaptureError(error)) throw error; + } + }, + }; +} + type SessionResourceTeardownRequest = { ref: SessionRef; sessionStore: SessionStore; diff --git a/packages/platform-apple/src/logs/start.test.ts b/packages/platform-apple/src/logs/start.test.ts index 7064190838..35f2074a78 100644 --- a/packages/platform-apple/src/logs/start.test.ts +++ b/packages/platform-apple/src/logs/start.test.ts @@ -4,6 +4,39 @@ import { expect, test, vi } from 'vitest'; import { startAppleAppLogs } from './start.ts'; import { appleDevice, hostFixture } from './runtime.fixtures.ts'; +test('a host log handle retries failed output disposal after its process exited', async () => { + const dispose = vi.fn(async () => {}).mockRejectedValueOnce(new Error('Close unavailable')); + const terminate = vi.fn(async () => {}); + const disposeProcess = vi.fn(async () => {}); + const fixture = hostFixture({ + startProcess: async () => ({ + wait: Promise.resolve({ stdout: '', stderr: '', exitCode: 0 }), + terminate, + [Symbol.asyncDispose]: disposeProcess, + }), + }); + const host = { + ...fixture.host, + outputs: { + ...fixture.host.outputs, + openAppend: async () => ({ write: async () => {}, [Symbol.asyncDispose]: dispose }), + }, + }; + const started = await startAppleAppLogs( + host, + appleDevice({ appleOs: 'macos', kind: 'device' }), + startInput(), + localRuntimeOwner('apple'), + ); + const handle = started.pendingHandle.transfer(); + await expect(handle.finish()).resolves.toMatchObject({ status: 'cleanup-pending' }); + await handle[Symbol.asyncDispose](); + await handle[Symbol.asyncDispose](); + expect(dispose).toHaveBeenCalledTimes(2); + expect(terminate).toHaveBeenCalledOnce(); + expect(disposeProcess).toHaveBeenCalledOnce(); +}); + test('simulator start resolves its app container through appleTools and keeps plutil generic', async () => { const appleToolRun = vi.fn(async () => ({ stdout: '/tmp/Example.app\n', diff --git a/packages/platform-apple/src/logs/start.ts b/packages/platform-apple/src/logs/start.ts index 1460a25270..212cff6682 100644 --- a/packages/platform-apple/src/logs/start.ts +++ b/packages/platform-apple/src/logs/start.ts @@ -278,18 +278,20 @@ function createAppleProcessHandle( state = 'failed'; }, ); + const cleanupSteps = new Set([ + async () => await backgroundProcess.terminate(), + async () => await backgroundProcess.wait.then(() => undefined), + async () => await backgroundProcess[Symbol.asyncDispose](), + async () => await output[Symbol.asyncDispose](), + ]); let finishPromise: | Promise> | undefined; const finish = async () => (finishPromise ??= (async () => { - const failures = await cleanupInOrder([ - async () => await backgroundProcess.terminate(), - async () => await backgroundProcess.wait.then(() => undefined), - async () => await backgroundProcess[Symbol.asyncDispose](), - async () => await output[Symbol.asyncDispose](), - ]); + const failures = await cleanupInOrder(cleanupSteps); if (failures.length > 0) { + finishPromise = undefined; state = 'failed'; return { status: 'cleanup-pending', @@ -309,11 +311,12 @@ function createAppleProcessHandle( }); } -async function cleanupInOrder(steps: readonly (() => Promise)[]): Promise { +async function cleanupInOrder(steps: Set<() => Promise>): Promise { const failures: unknown[] = []; for (const step of steps) { try { await step(); + steps.delete(step); } catch (error) { failures.push(error); } diff --git a/src/platform-runtime-perf-capture-host.test.ts b/src/platform-runtime-perf-capture-host.test.ts index f87592ceef..d55c777204 100644 --- a/src/platform-runtime-perf-capture-host.test.ts +++ b/src/platform-runtime-perf-capture-host.test.ts @@ -156,6 +156,25 @@ test('starts, retargets, finishes, and cleans a Simpleperf profile through one l expect(androidNative.cleanupAndroidNativePerfSession).toHaveBeenCalledTimes(1); }); +test('native perf disposal retries a failed cleanup and memoizes confirmation', async () => { + androidNative.startAndroidPerfettoTrace.mockResolvedValue( + androidCapture({ kind: 'perfetto', type: 'trace' }), + ); + androidNative.cleanupAndroidNativePerfSession.mockRejectedValueOnce( + new Error('Cleanup unavailable'), + ); + const started = await startAndroidPerfCapture( + device, + localRuntimeOwner('android'), + captureInput({ kind: 'perfetto', mode: 'trace' }), + ); + const handle = started.pendingHandle.transfer(); + await expect(handle[Symbol.asyncDispose]()).rejects.toThrow('Cleanup unavailable'); + await handle[Symbol.asyncDispose](); + await handle[Symbol.asyncDispose](); + expect(androidNative.cleanupAndroidNativePerfSession).toHaveBeenCalledTimes(2); +}); + test('starts and finishes an Android Perfetto trace', async () => { const capture = androidCapture({ kind: 'perfetto', type: 'trace' }); androidNative.startAndroidPerfettoTrace.mockResolvedValue(capture); diff --git a/src/platform-runtime-perf-capture-host.ts b/src/platform-runtime-perf-capture-host.ts index 0c1c360ec0..3c0b19fd12 100644 --- a/src/platform-runtime-perf-capture-host.ts +++ b/src/platform-runtime-perf-capture-host.ts @@ -13,6 +13,7 @@ import type { RuntimeOwnerRef } from '@agent-device/contracts/platform-runtime'; import type { ManagedProcessIdentity } from '@agent-device/contracts/platform-runtime-host'; import { createDurableResourceEnvelope, + createRetryableCleanup, decodeDurableDescriptor, encodeDurableDescriptor, } from '@agent-device/capture-kit'; @@ -179,7 +180,6 @@ function createCaptureHandle( ): PerfNativeCaptureLiveHandle { let outPath = steps.initialOutPath; let finish: Promise> | undefined; - let cleanup: Promise | undefined; const clean = async (): Promise => await steps.cleanup().then(() => ({ status: 'cleaned' }) as const, cleanupPending); return Object.freeze({ @@ -197,11 +197,7 @@ function createCaptureHandle( throw error; }, )), - forceCleanup: async () => (cleanup ??= clean()), - [Symbol.asyncDispose]: async () => { - const outcome = await (cleanup ??= clean()); - assertCleanup(outcome); - }, + ...createRetryableCleanup(clean, assertCleanup), }); }