From be20b30cee2477b7810ad6ee2cf05f94de744f94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Sun, 1 Nov 2026 11:01:33 +0100 Subject: [PATCH 1/9] fix(capture-kit): make a completed durable record terminal Fence equality proves ownership, not admission. Two callers that captured the same live resource present the same fence, so serialization alone let a queued cleanup or a late progress write drive a record another attempt had already committed back to `open`, and a cleanup-pending outcome could leave it open with its export already served. The fence owner now revalidates the record's lifecycle while it holds the fence. A completed record is terminal: a run that finds one is answered through `onCompleted` before it receives a lease, so it cannot repeat native or destructive work, and a transition that would reopen a committed record is refused with the typed reason `resource-already-completed`. Recovery's copy of `markCompleting` is gone; the one transition rule lives in `durable-capture/fence.ts`. --- .../src/durable-capture/fence.test.ts | 76 +++++++++++++ .../capture-kit/src/durable-capture/fence.ts | 45 +++++++- .../src/durable-capture/finish-recovered.ts | 6 ++ .../src/durable-capture/recovery.ts | 21 ++-- .../transitions-terminal.test.ts | 101 ++++++++++++++++++ .../src/durable-capture/transitions.ts | 43 ++++++-- 6 files changed, 270 insertions(+), 22 deletions(-) create mode 100644 packages/capture-kit/src/durable-capture/transitions-terminal.test.ts diff --git a/packages/capture-kit/src/durable-capture/fence.test.ts b/packages/capture-kit/src/durable-capture/fence.test.ts index 81c753cd00..56ed1826e0 100644 --- a/packages/capture-kit/src/durable-capture/fence.test.ts +++ b/packages/capture-kit/src/durable-capture/fence.test.ts @@ -20,6 +20,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 +42,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 +55,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 +71,79 @@ 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 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..1d7c7a932d 100644 --- a/packages/capture-kit/src/durable-capture/fence.ts +++ b/packages/capture-kit/src/durable-capture/fence.ts @@ -8,7 +8,8 @@ import type { 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,13 +20,30 @@ 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. A + * transition inside a run that committed its record is refused the same way, which is what keeps a + * late progress write from reopening an export that is already served. + */ 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 () => { @@ -44,12 +62,16 @@ export async function withDurableCaptureResourceFence( `${params.store.displayName} resource ownership fence was lost`, ); } + if (record.envelope.lifecycle === 'completed') { + return await params.onCompleted({ envelope: record.envelope }); + } let current = record.envelope; return await params.run({ get envelope() { return current; }, transition(lifecycle, update = {}) { + if (current.lifecycle === 'completed') throw completedDurableCaptureError(params.store); current = Object.freeze({ ...current, lifecycle, @@ -94,3 +116,20 @@ 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 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.ts b/packages/capture-kit/src/durable-capture/finish-recovered.ts index 0f00a619da..a6348d4ad2 100644 --- a/packages/capture-kit/src/durable-capture/finish-recovered.ts +++ b/packages/capture-kit/src/durable-capture/finish-recovered.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. Its committed export + // is what the caller is served, and this attempt adds nothing to it. + onCompleted: () => undefined, run: async (lease) => { transitionFinishOutcome(definition, lease, { status: 'completed', @@ -104,6 +107,9 @@ export async function finishRecoveredDurableCapture< store: definition.store, resourcePath: params.resourcePath, expected: envelope.fence, + // A committed record keeps its facts: this descriptor cleanup cannot make it anything the + // finish did not already record, and the caller still reports what cleanup returned. + onCompleted: () => undefined, run: async (lease) => transitionCleanupOutcome(lease, cleanup), }); requireConfirmedDurableCaptureCleanup(definition, cleanup); diff --git a/packages/capture-kit/src/durable-capture/recovery.ts b/packages/capture-kit/src/durable-capture/recovery.ts index 4f09648e0a..65cba8613e 100644 --- a/packages/capture-kit/src/durable-capture/recovery.ts +++ b/packages/capture-kit/src/durable-capture/recovery.ts @@ -13,10 +13,10 @@ 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 { markCompletingDurableCapture, transitionCleanupOutcome } from './transitions.ts'; const DEFAULT_RECOVERY_DEADLINE_MS = 5_000; @@ -162,8 +162,11 @@ async function settleRecoveryAuthority ({ status: 'already-missing' as const }), run: async (lease) => { - markCompleting(lease); + markCompletingDurableCapture(lease); const outcome = await handle.forceCleanup(); transitionCleanupOutcome(lease, outcome); return outcome; @@ -194,6 +197,9 @@ async function settleRecoveryAuthority undefined, run: async (lease) => transitionCleanupOutcome(lease, cleanup), }); 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..b48eac6d51 --- /dev/null +++ b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts @@ -0,0 +1,101 @@ +import { expect, test } from 'vitest'; +import { adoptStartedDurableCapture } from './adoption.ts'; +import { + createTestCaptureDefinition, + makeDurableCaptureContext, + makeDurableCaptureStartResult, + testCaptureDefinition, + testCaptureStore, +} from './durable-capture.fixtures.ts'; +import { forceCleanupLiveDurableCapture, finishLiveDurableCapture } from './transitions.ts'; + +/** + * 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 or dispose a committed export. + */ +test('a cleanup queued behind a committed finish leaves the record completed and cleans nothing', async () => { + const context = makeDurableCaptureContext(); + const start = makeDurableCaptureStartResult(context, { + cleanup: { status: 'cleanup-pending', reason: 'cleanup-unconfirmed' }, + }); + 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).not.toHaveBeenCalled(); + expect(testCaptureStore.read(context.resourcePath)).toMatchObject({ + status: 'decoded', + envelope: { lifecycle: 'completed', metadata: { phase: 'completed' } }, + }); +}); + +/** + * 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..5bcf886939 100644 --- a/packages/capture-kit/src/durable-capture/transitions.ts +++ b/packages/capture-kit/src/durable-capture/transitions.ts @@ -6,9 +6,14 @@ 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 { 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, @@ -56,7 +61,7 @@ export async function finishDurableCaptureHandle< definition: DurableCaptureRecordDefinition, params: { handle: H; - fence: DurableCaptureResourceFenceLease['envelope']['fence']; + fence: ResourceOwnershipFence; resourcePath: string; intent: DurableCaptureFinishIntent; }, @@ -65,8 +70,14 @@ export async function finishDurableCaptureHandle< store: definition.store, resourcePath: params.resourcePath, expected: params.fence, + // The caller that committed this record already returned its export. A finish arriving after + // that replays nothing of its own: the kind's completion is its own metadata, and the durable + // manifest is what serves it. + onCompleted: () => { + throw completedDurableCaptureError(definition.store); + }, run: async (lease) => { - markCompleting(lease); + markCompletingDurableCapture(lease); const progress = createStopProgress(lease); let finishOutcome: FinishOutcome; try { @@ -89,7 +100,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 +142,7 @@ async function respondToFailedFinish( - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, cleanupError: unknown, ): unknown | undefined { try { @@ -199,8 +210,13 @@ export async function forceCleanupLiveDurableCapture< store: definition.store, resourcePath: params.resourcePath, expected: active.envelope.fence, + // A committed record has already been through the finish that disposes what this kind's cleanup + // would: its own handle answers a cleanup after a completed finish with the same confirmed + // outcome without signalling anything. Settling the slot here is what releases the resource a + // caller captured before that finish, and leaves the record as committed. + onCompleted: () => ({ status: 'already-missing' as const }), run: async (lease) => { - markCompleting(lease); + markCompletingDurableCapture(lease); const result = await active.handle.forceCleanup(); transitionCleanupOutcome(lease, result); return result; @@ -211,7 +227,7 @@ export async function forceCleanupLiveDurableCapture< } export function transitionCleanupOutcome( - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, outcome: CleanupOutcome, ): void { lease.transition(isConfirmedCleanup(outcome) ? 'completed' : 'open', { @@ -254,7 +270,7 @@ function cleanupPendingError( export function transitionFinishOutcome( definition: DurableCaptureRecordDefinition, - lease: DurableCaptureResourceFenceLease, + lease: DurableCaptureOpenResourceFenceLease, outcome: FinishOutcome, ): void { if (outcome.status === 'completed') { @@ -277,7 +293,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 +303,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' }, }); From 83d7b4d7b143ce683086f5334240a704a33c48ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Sun, 1 Nov 2026 14:07:56 +0100 Subject: [PATCH 2/9] fix(capture-kit): settle a committed durable record for its queued callers Revalidate the persisted lifecycle at every transition, not only at lease issue, and revoke a lease when the run it was issued to ends, so an escaped progress or completion write cannot publish onto a record another run committed. Move descriptor cleanup inside the fence that decides terminal admission, at both recovery entry points, so destructive native cleanup cannot run for a record that committed while recovery authority was being acquired. Answer a finish that arrives after the commit with the completion the record holds, through a kind-owned optional reader, instead of refusing it. Session teardown treats the remaining refusal as settled rather than as an unreleased resource, so a second record stop stays what it was: a stop. --- .../screen-recording-session-resource.test.ts | 78 ++++++++++++ .../app-log-durable-resource.ts | 39 ++++++ .../screen-recording-resource-store.ts | 7 ++ .../screen-recording-session-resource.ts | 18 +-- .../screen-recording-stop-recovery.ts | 18 +-- .../src/durable-capture/definition.ts | 13 ++ .../durable-capture.fixtures.ts | 22 +++- .../src/durable-capture/fence.test.ts | 82 ++++++++++++- .../capture-kit/src/durable-capture/fence.ts | 116 ++++++++++++------ .../durable-capture/finish-recovered.test.ts | 83 +++++++++++++ .../src/durable-capture/finish-recovered.ts | 17 +-- .../capture-kit/src/durable-capture/index.ts | 1 + .../src/durable-capture/recovery.test.ts | 71 +++++++++++ .../src/durable-capture/recovery.ts | 24 ++-- .../transitions-terminal.test.ts | 85 ++++++++++++- .../src/durable-capture/transitions.ts | 53 ++++++-- .../session-teardown-resource-order.test.ts | 32 ++++- .../daemon/src/sessions/session-teardown.ts | 27 +++- 18 files changed, 690 insertions(+), 96 deletions(-) create mode 100644 packages/capture-kit/src/durable-capture/finish-recovered.test.ts 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..277e4e2869 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 @@ -273,6 +273,84 @@ 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 sessionStore = makeCaptureAdmissionSessionStore( + 'screen-recording-replayed-stop-', + ); + 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 signals = vi.fn(async () => ({ observation: { recorder: 'confirmed' as const } })); + const handle = createScreenRecordingLiveHandle( + { + backend: 'android', + outPath: '/tmp/recording.mp4', + 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' }), + }, + ); + await adoptStartedScreenRecording({ + admissionLedger: createScreenRecordingAdmissionLedger(), + binding, + device: session.device, + owner, + fence, + pendingHandle: new PendingTransferGuard(handle), + 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'); + + const stopped = await finishLiveScreenRecording({ intent: 'capture', binding }); + binding.adopt(queued); + + await expect(finishLiveScreenRecording({ intent: 'disposal', binding })).resolves.toEqual( + stopped, + ); + expect(signals).toHaveBeenCalledOnce(); + expect(binding.read()).toBeUndefined(); +}); + function resourcePath( sessionStore: CaptureAdmissionSessionStore, sessionName: string, 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..d5f00d9720 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')).readStoredCompletion(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..34e3de79df 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. @@ -94,11 +94,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( @@ -123,11 +122,14 @@ 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 + * has already bounded this metadata to frozen plain JSON, so what a 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. + * + * Two callers need exactly that verdict: a stop whose response never reached its caller, and a finish + * that reached the record after another caller committed it. Both are answered from here. */ -function readStoredCompletion( +export function readStoredCompletion( metadata: JsonObject | undefined, ): ScreenRecordingCompletion | undefined { const stored = metadata?.[SCREEN_RECORDING_COMPLETION_METADATA_KEY]; 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 56ed1826e0..9b120307e3 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({ @@ -117,6 +120,83 @@ test('a caller holding the same fence cannot write a record an earlier caller co }); }); +/** + * A progress port its own run kept holding outlives that run, and a later run commits the record it + * was issued against. Two leases, not two transitions on one lease: the escaped one must be dead the + * moment its own run ends, or its cached `open` view overwrites what the later run committed. + */ +test('a lease that outlived its run cannot write a record a later run committed', 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) => { + 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(); diff --git a/packages/capture-kit/src/durable-capture/fence.ts b/packages/capture-kit/src/durable-capture/fence.ts index 1d7c7a932d..dbb37585a9 100644 --- a/packages/capture-kit/src/durable-capture/fence.ts +++ b/packages/capture-kit/src/durable-capture/fence.ts @@ -4,7 +4,7 @@ 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>(); @@ -32,9 +32,13 @@ export type DurableCaptureCompletedResourceFenceLease = Readon * 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. A - * transition inside a run that committed its record is refused the same way, which is what keeps a - * late progress write from reopening an export that is already served. + * 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<{ @@ -47,44 +51,66 @@ export async function withDurableCaptureResourceFence( }>, ): 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}`, - ); - } - if (!sameFence(record.envelope.fence, params.expected)) { - throw fenceError( - params.store, - `${params.store.displayName} resource ownership fence was lost`, - ); + const admitted = requireAdmissibleRecord(params.store, params.resourcePath, params.expected); + if (admitted.envelope.lifecycle === 'completed') { + return await params.onCompleted({ envelope: admitted.envelope }); } - if (record.envelope.lifecycle === 'completed') { - return await params.onCompleted({ envelope: record.envelope }); + 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 = {}) { - if (current.lifecycle === 'completed') throw completedDurableCaptureError(params.store); - 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; } @@ -117,6 +143,22 @@ function fenceError(store: DurableCaptureResourceStore, message: string) }); } +/** + * 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 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 a6348d4ad2..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,8 +78,8 @@ export async function finishRecoveredDurableCapture< store: definition.store, resourcePath: params.resourcePath, expected: envelope.fence, - // The record committed while this recovery was acquiring authority. Its committed export - // is what the caller is served, and this attempt adds nothing to it. + // 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, { @@ -102,15 +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, - // A committed record keeps its facts: this descriptor cleanup cannot make it anything the - // finish did not already record, and the caller still reports what cleanup returned. - onCompleted: () => undefined, - 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 65cba8613e..821841fced 100644 --- a/packages/capture-kit/src/durable-capture/recovery.ts +++ b/packages/capture-kit/src/durable-capture/recovery.ts @@ -16,7 +16,11 @@ import { import { withDurableCaptureResourceFence } from './fence.ts'; import type { DurableCaptureRecordDefinition } from './definition.ts'; import { durableCaptureDiagnosticPrefix } from './labels.ts'; -import { markCompletingDurableCapture, transitionCleanupOutcome } from './transitions.ts'; +import { + cleanupDurableCaptureDescriptor, + markCompletingDurableCapture, + transitionCleanupOutcome, +} from './transitions.ts'; const DEFAULT_RECOVERY_DEADLINE_MS = 5_000; @@ -162,9 +166,10 @@ async function settleRecoveryAuthority ({ status: 'already-missing' as const }), + // The record committed between selection and this fence. The handle recovery reattached is + // still this process's native resource, so the terminal branch signals it too; only the + // record's own facts are off limits to it. + onCompleted: async () => await handle.forceCleanup(), run: async (lease) => { markCompletingDurableCapture(lease); const outcome = await handle.forceCleanup(); @@ -192,15 +197,10 @@ async function settleRecoveryAuthority undefined, - run: async (lease) => transitionCleanupOutcome(lease, cleanup), + cleanup: () => authority.control.cleanup(envelope), }); return isConfirmedCleanup(cleanup); } diff --git a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts index b48eac6d51..b2e862d6d2 100644 --- a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts +++ b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts @@ -12,13 +12,12 @@ import { forceCleanupLiveDurableCapture, finishLiveDurableCapture } from './tran /** * 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 or dispose a committed export. + * 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 cleans nothing', async () => { +test('a cleanup queued behind a committed finish leaves the record completed and settles its handle', async () => { const context = makeDurableCaptureContext(); - const start = makeDurableCaptureStartResult(context, { - cleanup: { status: 'cleanup-pending', reason: 'cleanup-unconfirmed' }, - }); + const start = makeDurableCaptureStartResult(context); await adoptStartedDurableCapture( testCaptureDefinition, { ...context, ...start, throwIfCanceled: () => {} }, @@ -45,11 +44,85 @@ test('a cleanup queued behind a committed finish leaves the record completed and resourcePath: context.resourcePath, }); - expect(start.forceCleanup).not.toHaveBeenCalled(); + 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(); }); /** diff --git a/packages/capture-kit/src/durable-capture/transitions.ts b/packages/capture-kit/src/durable-capture/transitions.ts index 5bcf886939..d302706417 100644 --- a/packages/capture-kit/src/durable-capture/transitions.ts +++ b/packages/capture-kit/src/durable-capture/transitions.ts @@ -7,6 +7,7 @@ import { } 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 { @@ -70,10 +71,13 @@ export async function finishDurableCaptureHandle< store: definition.store, resourcePath: params.resourcePath, expected: params.fence, - // The caller that committed this record already returned its export. A finish arriving after - // that replays nothing of its own: the kind's completion is its own metadata, and the durable - // manifest is what serves it. - onCompleted: () => { + // 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 }) => { + const replayed = await definition.completionFromMetadata?.(envelope.metadata); + if (replayed !== undefined) return replayed; throw completedDurableCaptureError(definition.store); }, run: async (lease) => { @@ -210,11 +214,11 @@ export async function forceCleanupLiveDurableCapture< store: definition.store, resourcePath: params.resourcePath, expected: active.envelope.fence, - // A committed record has already been through the finish that disposes what this kind's cleanup - // would: its own handle answers a cleanup after a completed finish with the same confirmed - // outcome without signalling anything. Settling the slot here is what releases the resource a - // caller captured before that finish, and leaves the record as committed. - onCompleted: () => ({ status: 'already-missing' as const }), + // 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) => { markCompletingDurableCapture(lease); const result = await active.handle.forceCleanup(); @@ -226,6 +230,37 @@ 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: DurableCaptureOpenResourceFenceLease, outcome: CleanupOutcome, 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..9085601df8 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,19 @@ 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 ({ run }) => { + stopEvents.terminal.add('audio_probe'); + + await expect(run({ sessionName: 'stop-terminal-session' })).resolves.not.toThrow(); + + 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; From daf7555f189dcca77ac5cb6a2270ac23dffdc2fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Sun, 1 Nov 2026 15:38:46 +0100 Subject: [PATCH 3/9] test(capture-kit): cover app-log terminal replay and its refusals The Coverage lane scored the new app-log completion reader CRAP 110 at 0% coverage: nothing exercised it. These cases drive it through the resource it belongs to, so a queued teardown finish is answered from the record the first stop committed, and a record whose journal cannot name its whole completion keeps the typed refusal instead of inventing a log path. --- .../app-log-durable-resource-replay.test.ts | 198 ++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 packages/capture-kit/src/capture-admission/app-log-durable-resource-replay.test.ts 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(); + }, +); From ce5d76e829be77eaec346e05924a2dfa90f8eb48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Sun, 1 Nov 2026 16:17:37 +0100 Subject: [PATCH 4/9] fix(capture-kit): gate replayed recordings on the surviving export A finish that reached a committed screen-recording record was answered with the stored stop response whatever had since happened to the clip, so a deleted export came back as a successful stop naming a path nothing can download. The surviving-file check now belongs to the shared completion reader, the one owner both the stop-recovery route and the terminal replay answer from; a gone export keeps the typed already-completed refusal that teardown settles. --- .../screen-recording-session-resource.test.ts | 68 +++++++++++++++---- .../screen-recording-stop-recovery.ts | 16 +++-- 2 files changed, 63 insertions(+), 21 deletions(-) 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 277e4e2869..e0d2e14016 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 { @@ -281,9 +284,44 @@ test('a record stop that fails after collecting resumes through the fence withou * 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 sessionStore = makeCaptureAdmissionSessionStore( - 'screen-recording-replayed-stop-', - ); + 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.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); @@ -293,11 +331,13 @@ test('a disposal queued behind a committed recording stop replays that export', }); 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: '/tmp/recording.mp4', + outPath, startedAt: 1, scope: 'app', showTouches: false, @@ -340,16 +380,16 @@ test('a disposal queued behind a committed recording stop replays that export', // 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'); - - const stopped = await finishLiveScreenRecording({ intent: 'capture', binding }); - binding.adopt(queued); - - await expect(finishLiveScreenRecording({ intent: 'disposal', binding })).resolves.toEqual( - stopped, - ); - expect(signals).toHaveBeenCalledOnce(); - expect(binding.read()).toBeUndefined(); -}); + return { + binding, + signals, + outPath, + /** Puts the slot back the way the queued teardown left it holding it. */ + requeue: () => { + binding.adopt(queued); + }, + }; +} function resourcePath( sessionStore: CaptureAdmissionSessionStore, 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 34e3de79df..9b280fe8cf 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 @@ -65,7 +65,6 @@ export function resolveScreenRecordingStopRecovery( }); return { kind: 'none' }; } - if (!fs.existsSync(completion.outPath)) return { kind: 'none' }; return { kind: 'completed', completion }; } @@ -121,13 +120,15 @@ 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 a reader 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. * - * Two callers need exactly that verdict: a stop whose response never reached its caller, and a finish - * that reached the record after another caller committed it. Both are answered from here. + * 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. */ export function readStoredCompletion( metadata: JsonObject | undefined, @@ -141,7 +142,8 @@ export 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. */ From 5221af1c26c7652acffe1d5ac6803525740636b2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Sun, 1 Nov 2026 22:12:29 +0100 Subject: [PATCH 5/9] test(capture): pin terminal replay and released lease admission --- .../screen-recording-session-resource.test.ts | 5 ++++- .../src/durable-capture/fence.test.ts | 20 +++++++++++++++---- .../session-teardown-resource-order.test.ts | 6 ++++-- 3 files changed, 24 insertions(+), 7 deletions(-) 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 e0d2e14016..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 @@ -294,6 +294,7 @@ test('a disposal queued behind a committed recording stop replays that export', await expect( finishLiveScreenRecording({ intent: 'disposal', binding: recording.binding }), ).resolves.toEqual(stopped); + expect(recording.finish).toHaveBeenCalledOnce(); expect(recording.signals).toHaveBeenCalledOnce(); expect(recording.binding.read()).toBeUndefined(); }); @@ -359,13 +360,14 @@ async function adoptCommittedRecording(prefix: string) { forceCleanup: async () => ({ status: 'cleaned' }), }, ); + const finish = vi.fn(handle.finish); await adoptStartedScreenRecording({ admissionLedger: createScreenRecordingAdmissionLedger(), binding, device: session.device, owner, fence, - pendingHandle: new PendingTransferGuard(handle), + pendingHandle: new PendingTransferGuard({ ...handle, finish }), envelope: createDurableResourceEnvelope({ resourceKind: 'screen-recording', sessionId: sessionName, @@ -382,6 +384,7 @@ async function adoptCommittedRecording(prefix: string) { 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. */ diff --git a/packages/capture-kit/src/durable-capture/fence.test.ts b/packages/capture-kit/src/durable-capture/fence.test.ts index 9b120307e3..acc0ffe112 100644 --- a/packages/capture-kit/src/durable-capture/fence.test.ts +++ b/packages/capture-kit/src/durable-capture/fence.test.ts @@ -121,11 +121,10 @@ test('a caller holding the same fence cannot write a record an earlier caller co }); /** - * A progress port its own run kept holding outlives that run, and a later run commits the record it - * was issued against. Two leases, not two transitions on one lease: the escaped one must be dead the - * moment its own run ends, or its cached `open` view overwrites what the later run committed. + * 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 that outlived its run cannot write a record a later run committed', async () => { +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({ @@ -144,6 +143,19 @@ test('a lease that outlived its run cannot write a record a later run committed' 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' } }); }, }); 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 9085601df8..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 @@ -230,10 +230,12 @@ test('daemon teardown omits the app-log stop once the app-log fence has settled' */ test.each(CALLERS)( '$name settles a resource whose record already committed instead of reporting it unreleased', - async ({ run }) => { + async ({ name, run }) => { stopEvents.terminal.add('audio_probe'); - await expect(run({ sessionName: 'stop-terminal-session' })).resolves.not.toThrow(); + 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); }, From 7246214dc0e38695456c1683e76e90a7940a0eb7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Sun, 1 Nov 2026 23:22:24 +0100 Subject: [PATCH 6/9] fix(capture): dispose captured handles before terminal replay --- .../screen-recording-session-resource.ts | 2 +- .../screen-recording-stop-recovery.ts | 4 +- .../transitions-terminal.test.ts | 80 ++++++++++++++++++- .../src/durable-capture/transitions.ts | 1 + 4 files changed, 82 insertions(+), 5 deletions(-) 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 d5f00d9720..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 @@ -40,7 +40,7 @@ export const screenRecordingDurableResource = createDurableCaptureResource< // 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')).readStoredCompletion(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', 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 9b280fe8cf..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 @@ -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', @@ -130,7 +130,7 @@ function assertManifestBelongsToRequest( * 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. */ -export function readStoredCompletion( +export function readReplayableStoredCompletion( metadata: JsonObject | undefined, ): ScreenRecordingCompletion | undefined { const stored = metadata?.[SCREEN_RECORDING_COMPLETION_METADATA_KEY]; diff --git a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts index b2e862d6d2..81773ca88e 100644 --- a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts +++ b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts @@ -1,4 +1,4 @@ -import { expect, test } from 'vitest'; +import { expect, test, vi } from 'vitest'; import { adoptStartedDurableCapture } from './adoption.ts'; import { createTestCaptureDefinition, @@ -7,7 +7,83 @@ import { testCaptureDefinition, testCaptureStore, } from './durable-capture.fixtures.ts'; -import { forceCleanupLiveDurableCapture, finishLiveDurableCapture } from './transitions.ts'; +import { + forceCleanupLiveDurableCapture, + finishLiveDurableCapture, + finishDurableCaptureHandle, +} from './transitions.ts'; + +test.each(['replay', 'refusal'] as const)( + 'a terminal %s disposes 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 dispose = vi.spyOn(captured.handle, Symbol.asyncDispose); + + 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(dispose).toHaveBeenCalledOnce(); + expect(captured.forceCleanup).toHaveBeenCalledOnce(); + expect(captured.finish).not.toHaveBeenCalled(); + expect(context.binding.read()).toBeUndefined(); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); + }, +); + +test('a terminal finish reports captured-handle disposal failure instead of a settled refusal', async () => { + const definition = { ...testCaptureDefinition, completionFromMetadata: undefined }; + const cleanupError = new Error('Captured handle disposal failed'); + const { context, captured, record } = await captureWhoseRecordAnotherHandleCommitted( + definition, + cleanupError, + ); + + await expect( + finishLiveDurableCapture( + definition, + { binding: context.binding, intent: 'disposal' }, + context.resourcePath, + ), + ).rejects.toBe(cleanupError); + expect(captured.forceCleanup).toHaveBeenCalledOnce(); + expect(captured.finish).not.toHaveBeenCalled(); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); +}); + +async function captureWhoseRecordAnotherHandleCommitted( + definition: typeof testCaptureDefinition, + cleanupError?: Error, +) { + const context = makeDurableCaptureContext(); + const captured = makeDurableCaptureStartResult(context, { cleanupError }); + 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 diff --git a/packages/capture-kit/src/durable-capture/transitions.ts b/packages/capture-kit/src/durable-capture/transitions.ts index d302706417..9a5309a3d1 100644 --- a/packages/capture-kit/src/durable-capture/transitions.ts +++ b/packages/capture-kit/src/durable-capture/transitions.ts @@ -76,6 +76,7 @@ export async function finishDurableCaptureHandle< // 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 }) => { + await params.handle[Symbol.asyncDispose](); const replayed = await definition.completionFromMetadata?.(envelope.metadata); if (replayed !== undefined) return replayed; throw completedDurableCaptureError(definition.store); From 2d27d0755e45bb3236162dcef4fcaa652a60c015 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Mon, 2 Nov 2026 00:12:11 +0100 Subject: [PATCH 7/9] fix(capture): retain pending cleanup without losing committed replay --- .../capture-kit/src/app-log-live-handle.ts | 19 ++- .../src/app-log-pid-process.test.ts | 21 +++ .../capture-kit/src/app-log-pid-process.ts | 3 + .../src/audio-probe-runtime.test.ts | 42 ++++++ .../capture-kit/src/audio-probe-runtime.ts | 24 +++- .../transitions-terminal.test.ts | 125 +++++++++++++++--- .../src/durable-capture/transitions.ts | 45 ++++++- .../src/provider-app-log-poller.test.ts | 8 +- .../src/provider-app-log-poller.ts | 1 + .../src/screen-recording-live-handle.test.ts | 19 +++ .../src/screen-recording-live-handle.ts | 32 ++++- .../platform-apple/src/logs/start.test.ts | 29 ++++ packages/platform-apple/src/logs/start.ts | 1 + ...platform-runtime-perf-capture-host.test.ts | 19 +++ src/platform-runtime-perf-capture-host.ts | 9 +- 15 files changed, 356 insertions(+), 41 deletions(-) diff --git a/packages/capture-kit/src/app-log-live-handle.ts b/packages/capture-kit/src/app-log-live-handle.ts index 0e7e2f252d..7fb24153f6 100644 --- a/packages/capture-kit/src/app-log-live-handle.ts +++ b/packages/capture-kit/src/app-log-live-handle.ts @@ -23,13 +23,28 @@ export function createAppLogLiveHandle( let finish: Promise> | undefined; let cleanup: Promise | undefined; let disposal: Promise | undefined; - const forceCleanup = (): Promise => (cleanup ??= implementation.forceCleanup()); + const forceCleanup = (): Promise => + (cleanup ??= implementation.forceCleanup().then( + (outcome) => { + if (!isConfirmedCleanup(outcome)) cleanup = undefined; + return outcome; + }, + (error) => { + cleanup = undefined; + throw error; + }, + )); return Object.freeze({ inspect: () => implementation.inspect(), finish: () => (finish ??= implementation.finish()), forceCleanup, [Symbol.asyncDispose]: async () => { - disposal ??= forceCleanup().then(assertConfirmedCleanup); + disposal ??= forceCleanup() + .then(assertConfirmedCleanup) + .catch((error) => { + disposal = undefined; + throw error; + }); await disposal; }, }); 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..c48ef16015 100644 --- a/packages/capture-kit/src/app-log-pid-process.ts +++ b/packages/capture-kit/src/app-log-pid-process.ts @@ -98,6 +98,9 @@ export async function createPidScopedAppLogProcess( state = next; }, output, + }).then((outcome) => { + if (outcome.status === 'cleanup-pending') finishPromise = undefined; + return outcome; })); const startedAt = options.host.clock.now(); return createAppLogLiveHandleFromFinish({ 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..b4a1be2cd1 100644 --- a/packages/capture-kit/src/audio-probe-runtime.ts +++ b/packages/capture-kit/src/audio-probe-runtime.ts @@ -139,13 +139,24 @@ 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); + const forceCleanup = () => + (cleanup ??= clean().then( + (outcome) => { + if (!isConfirmedCleanup(outcome)) cleanup = undefined; + return outcome; + }, + (error) => { + cleanup = undefined; + throw error; + }, + )); return Object.freeze({ inspect: () => snapshot, status: async () => { @@ -157,7 +168,12 @@ function createHostAudioProbeLiveHandle( finish: finishProbe, forceCleanup, [Symbol.asyncDispose]: async () => { - disposal ??= forceCleanup().then(assertConfirmedAudioProbeCleanup); + disposal ??= forceCleanup() + .then(assertConfirmedAudioProbeCleanup) + .catch((error) => { + disposal = undefined; + throw error; + }); await disposal; }, }); diff --git a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts index 81773ca88e..336760750d 100644 --- a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts +++ b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts @@ -1,4 +1,12 @@ 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, @@ -6,6 +14,7 @@ import { makeDurableCaptureStartResult, testCaptureDefinition, testCaptureStore, + type TestCaptureHandle, } from './durable-capture.fixtures.ts'; import { forceCleanupLiveDurableCapture, @@ -44,32 +53,108 @@ test.each(['replay', 'refusal'] as const)( }, ); -test('a terminal finish reports captured-handle disposal failure instead of a settled refusal', async () => { - const definition = { ...testCaptureDefinition, completionFromMetadata: undefined }; - const cleanupError = new Error('Captured handle disposal failed'); - const { context, captured, record } = await captureWhoseRecordAnotherHandleCommitted( - definition, - cleanupError, - ); +test.each(['replay', 'refusal'] as const)( + 'a failed terminal disposal retains the %s binding 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 dispose = vi + .spyOn(captured.handle, Symbol.asyncDispose) + .mockRejectedValueOnce(new Error('Captured handle disposal failed')); + const logPath = `${context.resourcePath}.diagnostics.ndjson`; + const finish = () => + finishLiveDurableCapture( + definition, + { binding: context.binding, intent: 'disposal' }, + 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 disposal failed', + }, + }), + }), + ); + if (outcome === 'replay') await expect(finish()).resolves.toEqual(completion); + else + await expect(finish()).rejects.toMatchObject({ + details: { reason: 'resource-already-completed' }, + }); + expect(dispose).toHaveBeenCalledTimes(2); + expect(captured.finish).not.toHaveBeenCalled(); + expect(context.binding.read()).toBeUndefined(); + expect(testCaptureStore.read(context.resourcePath)).toEqual(record); + }, +); - await expect( - finishLiveDurableCapture( - definition, - { binding: context.binding, intent: 'disposal' }, - context.resourcePath, - ), - ).rejects.toBe(cleanupError); - expect(captured.forceCleanup).toHaveBeenCalledOnce(); - expect(captured.finish).not.toHaveBeenCalled(); - 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: 'disposal' }, + 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, - cleanupError?: Error, + handle?: TestCaptureHandle, ) { const context = makeDurableCaptureContext(); - const captured = makeDurableCaptureStartResult(context, { cleanupError }); + const start = makeDurableCaptureStartResult(context); + const captured = handle + ? { ...start, handle, pendingHandle: new PendingTransferGuard(handle) } + : start; await adoptStartedDurableCapture( definition, { ...context, ...captured, throwIfCanceled: () => {} }, diff --git a/packages/capture-kit/src/durable-capture/transitions.ts b/packages/capture-kit/src/durable-capture/transitions.ts index 9a5309a3d1..bf10ec7cfc 100644 --- a/packages/capture-kit/src/durable-capture/transitions.ts +++ b/packages/capture-kit/src/durable-capture/transitions.ts @@ -1,5 +1,6 @@ import { type CleanupOutcome, + type CleanupPendingReason, type DurableCaptureProgress, type FinishOutcome, type LiveResourceHandle, @@ -22,6 +23,14 @@ import type { } from './definition.ts'; import { capitalizeDurableCaptureLabel, durableCaptureDiagnosticPrefix } from './labels.ts'; +const CLEANUP_PENDING_REASONS: Readonly> = { + 'ownership-fence-lost': true, + 'owner-unavailable': true, + 'transport-failed': true, + 'cleanup-unconfirmed': true, + 'manual-recovery-required': true, +}; + export async function finishLiveDurableCapture< K extends string, H extends LiveResourceHandle, @@ -36,18 +45,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; @@ -65,6 +82,7 @@ export async function finishDurableCaptureHandle< fence: ResourceOwnershipFence; resourcePath: string; intent: DurableCaptureFinishIntent; + onCleanupPending?(outcome: Extract): void; }, ): Promise { return await withDurableCaptureResourceFence({ @@ -76,9 +94,29 @@ export async function finishDurableCaptureHandle< // 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 }) => { - await params.handle[Symbol.asyncDispose](); + let cleanup: Extract | undefined; + try { + await params.handle[Symbol.asyncDispose](); + } catch (error) { + const reason = error instanceof AppError ? error.details?.reason : undefined; + cleanup = { + status: 'cleanup-pending', + reason: + typeof reason === 'string' && Object.hasOwn(CLEANUP_PENDING_REASONS, reason) + ? (reason as CleanupPendingReason) + : 'cleanup-unconfirmed', + message: errorMessage(error), + }; + params.onCleanupPending?.(cleanup); + emitDiagnostic({ + level: 'warn', + phase: `${durableCaptureDiagnosticPrefix(definition.resourceKind)}_terminal_cleanup_pending`, + data: { resourcePath: params.resourcePath, cleanup }, + }); + } const replayed = await definition.completionFromMetadata?.(envelope.metadata); if (replayed !== undefined) return replayed; + if (cleanup) throw cleanupPendingError(definition, cleanup); throw completedDurableCaptureError(definition.store); }, run: async (lease) => { @@ -298,6 +336,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, }, 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..bf20e550aa 100644 --- a/packages/capture-kit/src/provider-app-log-poller.test.ts +++ b/packages/capture-kit/src/provider-app-log-poller.test.ts @@ -122,10 +122,9 @@ describe('provider app-log poller', () => { test('still disposes the output when reader cleanup rejects', async () => { const sleeps = deferredSleeps(); let outputDisposed = false; + 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, { @@ -146,6 +145,9 @@ 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); }); 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..9525069782 100644 --- a/packages/capture-kit/src/provider-app-log-poller.ts +++ b/packages/capture-kit/src/provider-app-log-poller.ts @@ -98,6 +98,7 @@ function createPollerHandle( await polling; const failures = await disposeAll([options.reader, output]); if (failures.length > 0) { + finishPromise = undefined; state = 'failed'; return { status: 'cleanup-pending', 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..cd61673b3b 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,25 @@ import assert from 'node:assert/strict'; import { 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 Error('Transport unavailable')); + else cleanup.mockResolvedValueOnce({ status: 'cleanup-pending', reason: 'transport-failed' }); + const finish = vi.fn(async () => completed()); + const handle = createScreenRecordingLiveHandle(snapshot(), { finish, forceCleanup: cleanup }); + await assert.rejects(async () => await handle[Symbol.asyncDispose]()); + 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..e6f830ab8a 100644 --- a/packages/capture-kit/src/screen-recording-live-handle.ts +++ b/packages/capture-kit/src/screen-recording-live-handle.ts @@ -34,23 +34,23 @@ export function createScreenRecordingLiveHandle( ): ScreenRecordingLiveHandle { let snapshot = freezeSnapshot(initial); let finish: Promise> | undefined; - let finishRequested = false; + let terminalRequested = false; let cleanup: Promise | undefined; let disposal: Promise | undefined; // 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 +58,20 @@ export function createScreenRecordingLiveHandle( : await implementation.forceCleanup(snapshot), async () => await implementation.forceCleanup(snapshot), ) - : implementation.forceCleanup(snapshot)); + : implementation.forceCleanup(snapshot); + const forceCleanup = () => { + terminalRequested = true; + return (cleanup ??= clean().then( + (outcome) => { + if (!isConfirmedCleanup(outcome)) cleanup = undefined; + return outcome; + }, + (error) => { + cleanup = undefined; + throw error; + }, + )); + }; return Object.freeze({ inspect: () => snapshot, appendGestureEvents: (events: readonly RecordingGestureEvent[]) => { @@ -86,7 +99,12 @@ export function createScreenRecordingLiveHandle( finish: finishRecording, forceCleanup, [Symbol.asyncDispose]: async () => { - disposal ??= forceCleanup().then(assertConfirmedCleanup); + disposal ??= forceCleanup() + .then(assertConfirmedCleanup) + .catch((error) => { + disposal = undefined; + throw error; + }); await disposal; }, }); diff --git a/packages/platform-apple/src/logs/start.test.ts b/packages/platform-apple/src/logs/start.test.ts index 7064190838..463c044881 100644 --- a/packages/platform-apple/src/logs/start.test.ts +++ b/packages/platform-apple/src/logs/start.test.ts @@ -4,6 +4,35 @@ 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 fixture = hostFixture({ + startProcess: async () => ({ + wait: Promise.resolve({ stdout: '', stderr: '', exitCode: 0 }), + terminate: async () => {}, + [Symbol.asyncDispose]: async () => {}, + }), + }); + 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); +}); + 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..eb6bc1f966 100644 --- a/packages/platform-apple/src/logs/start.ts +++ b/packages/platform-apple/src/logs/start.ts @@ -290,6 +290,7 @@ function createAppleProcessHandle( async () => await output[Symbol.asyncDispose](), ]); if (failures.length > 0) { + finishPromise = undefined; state = 'failed'; return { status: 'cleanup-pending', 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..5027ebb181 100644 --- a/src/platform-runtime-perf-capture-host.ts +++ b/src/platform-runtime-perf-capture-host.ts @@ -182,6 +182,11 @@ function createCaptureHandle( let cleanup: Promise | undefined; const clean = async (): Promise => await steps.cleanup().then(() => ({ status: 'cleaned' }) as const, cleanupPending); + const forceCleanup = async () => + (cleanup ??= clean().then((outcome) => { + if (outcome.status === 'cleanup-pending') cleanup = undefined; + return outcome; + })); return Object.freeze({ inspect: steps.inspect, setOutputPath: (value: string) => { @@ -197,9 +202,9 @@ function createCaptureHandle( throw error; }, )), - forceCleanup: async () => (cleanup ??= clean()), + forceCleanup, [Symbol.asyncDispose]: async () => { - const outcome = await (cleanup ??= clean()); + const outcome = await forceCleanup(); assertCleanup(outcome); }, }); From f4a801c9503b98b66670945cae4b13e66d8f0c1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Mon, 2 Nov 2026 07:50:45 +0100 Subject: [PATCH 8/9] refactor(capture): share cleanup retries and report pending disposal --- .../capture-kit/src/app-log-live-handle.ts | 25 +------ .../capture-kit/src/app-log-pid-process.ts | 48 +++++++------- .../capture-kit/src/audio-probe-runtime.ts | 25 +------ .../transitions-terminal.test.ts | 65 +++++++++++++++---- .../src/durable-capture/transitions.ts | 24 ++----- .../src/durable-resource-cleanup.ts | 24 +++++++ packages/capture-kit/src/index.ts | 1 + .../src/provider-app-log-poller.test.ts | 8 ++- .../src/provider-app-log-poller.ts | 10 ++- .../src/screen-recording-live-handle.test.ts | 13 +++- .../src/screen-recording-live-handle.ts | 30 ++------- .../platform-apple/src/logs/start.test.ts | 8 ++- packages/platform-apple/src/logs/start.ts | 16 +++-- src/platform-runtime-perf-capture-host.ts | 13 +--- 14 files changed, 154 insertions(+), 156 deletions(-) create mode 100644 packages/capture-kit/src/durable-resource-cleanup.ts diff --git a/packages/capture-kit/src/app-log-live-handle.ts b/packages/capture-kit/src/app-log-live-handle.ts index 7fb24153f6..1963f76aa8 100644 --- a/packages/capture-kit/src/app-log-live-handle.ts +++ b/packages/capture-kit/src/app-log-live-handle.ts @@ -1,3 +1,4 @@ +import { createRetryableCleanup } from './durable-resource-cleanup.ts'; import { AppError } from '@agent-device/kernel/errors'; import type { AppLogCompletion, @@ -21,32 +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().then( - (outcome) => { - if (!isConfirmedCleanup(outcome)) cleanup = undefined; - return outcome; - }, - (error) => { - cleanup = undefined; - throw error; - }, - )); return Object.freeze({ inspect: () => implementation.inspect(), finish: () => (finish ??= implementation.finish()), - forceCleanup, - [Symbol.asyncDispose]: async () => { - disposal ??= forceCleanup() - .then(assertConfirmedCleanup) - .catch((error) => { - disposal = undefined; - throw error; - }); - await disposal; - }, + ...createRetryableCleanup(() => implementation.forceCleanup(), assertConfirmedCleanup), }); } diff --git a/packages/capture-kit/src/app-log-pid-process.ts b/packages/capture-kit/src/app-log-pid-process.ts index c48ef16015..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,22 @@ 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; @@ -112,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 { @@ -144,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.ts b/packages/capture-kit/src/audio-probe-runtime.ts index b4a1be2cd1..f1da23b94e 100644 --- a/packages/capture-kit/src/audio-probe-runtime.ts +++ b/packages/capture-kit/src/audio-probe-runtime.ts @@ -1,3 +1,4 @@ +import { createRetryableCleanup } from './durable-resource-cleanup.ts'; import fs from 'node:fs/promises'; import { deviceIdentity, type DeviceInfo } from '@agent-device/kernel/device'; import { AppError } from '@agent-device/kernel/errors'; @@ -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 @@ -146,17 +145,6 @@ function createHostAudioProbeLiveHandle( async () => await terminateForCleanup(process), ) : terminateForCleanup(process); - const forceCleanup = () => - (cleanup ??= clean().then( - (outcome) => { - if (!isConfirmedCleanup(outcome)) cleanup = undefined; - return outcome; - }, - (error) => { - cleanup = undefined; - throw error; - }, - )); return Object.freeze({ inspect: () => snapshot, status: async () => { @@ -166,16 +154,7 @@ function createHostAudioProbeLiveHandle( return status ?? finalizeStatus(snapshot, undefined, 'not-started'); }, finish: finishProbe, - forceCleanup, - [Symbol.asyncDispose]: async () => { - disposal ??= forceCleanup() - .then(assertConfirmedAudioProbeCleanup) - .catch((error) => { - disposal = undefined; - throw error; - }); - await disposal; - }, + ...createRetryableCleanup(clean, assertConfirmedAudioProbeCleanup), }); } diff --git a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts index 336760750d..dc14f427b7 100644 --- a/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts +++ b/packages/capture-kit/src/durable-capture/transitions-terminal.test.ts @@ -23,7 +23,7 @@ import { } from './transitions.ts'; test.each(['replay', 'refusal'] as const)( - 'a terminal %s disposes the captured handle another handle did not release', + 'a terminal %s cleans the captured handle another handle did not release', async (outcome) => { const definition = { ...testCaptureDefinition, @@ -32,7 +32,6 @@ test.each(['replay', 'refusal'] as const)( }; const { context, captured, completion, record } = await captureWhoseRecordAnotherHandleCommitted(definition); - const dispose = vi.spyOn(captured.handle, Symbol.asyncDispose); const finish = finishLiveDurableCapture( definition, @@ -45,7 +44,6 @@ test.each(['replay', 'refusal'] as const)( details: { reason: 'resource-already-completed' }, }); - expect(dispose).toHaveBeenCalledOnce(); expect(captured.forceCleanup).toHaveBeenCalledOnce(); expect(captured.finish).not.toHaveBeenCalled(); expect(context.binding.read()).toBeUndefined(); @@ -54,7 +52,7 @@ test.each(['replay', 'refusal'] as const)( ); test.each(['replay', 'refusal'] as const)( - 'a failed terminal disposal retains the %s binding for a later successful teardown', + 'capture intent retains the %s binding after rejected cleanup for a later successful teardown', async (outcome) => { const definition = { ...testCaptureDefinition, @@ -63,14 +61,14 @@ test.each(['replay', 'refusal'] as const)( }; const { context, captured, completion, record } = await captureWhoseRecordAnotherHandleCommitted(definition); - const dispose = vi - .spyOn(captured.handle, Symbol.asyncDispose) - .mockRejectedValueOnce(new Error('Captured handle disposal failed')); + const cleanup = vi + .spyOn(captured.handle, 'forceCleanup') + .mockRejectedValueOnce(new Error('Captured handle cleanup failed')); const logPath = `${context.resourcePath}.diagnostics.ndjson`; - const finish = () => + const finish = (intent: 'capture' | 'disposal' = 'capture') => finishLiveDurableCapture( definition, - { binding: context.binding, intent: 'disposal' }, + { binding: context.binding, intent }, context.resourcePath, ); await withDiagnosticsScope({ command: 'record', logPath }, async () => { @@ -97,23 +95,62 @@ test.each(['replay', 'refusal'] as const)( cleanup: { status: 'cleanup-pending', reason: 'cleanup-unconfirmed', - message: 'Captured handle disposal failed', + message: 'Captured handle cleanup failed', }, }), }), ); - if (outcome === 'replay') await expect(finish()).resolves.toEqual(completion); + if (outcome === 'replay') await expect(finish('disposal')).resolves.toEqual(completion); else - await expect(finish()).rejects.toMatchObject({ + await expect(finish('disposal')).rejects.toMatchObject({ details: { reason: 'resource-already-completed' }, }); - expect(dispose).toHaveBeenCalledTimes(2); + 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) => { @@ -134,7 +171,7 @@ test.each(['rejected', 'unconfirmed'] as const)( const finish = () => finishLiveDurableCapture( testCaptureDefinition, - { binding: context.binding, intent: 'disposal' }, + { binding: context.binding, intent: 'capture' }, context.resourcePath, ); await expect(finish()).resolves.toEqual(completion); diff --git a/packages/capture-kit/src/durable-capture/transitions.ts b/packages/capture-kit/src/durable-capture/transitions.ts index bf10ec7cfc..6c82c9cef1 100644 --- a/packages/capture-kit/src/durable-capture/transitions.ts +++ b/packages/capture-kit/src/durable-capture/transitions.ts @@ -1,6 +1,5 @@ import { type CleanupOutcome, - type CleanupPendingReason, type DurableCaptureProgress, type FinishOutcome, type LiveResourceHandle, @@ -23,14 +22,6 @@ import type { } from './definition.ts'; import { capitalizeDurableCaptureLabel, durableCaptureDiagnosticPrefix } from './labels.ts'; -const CLEANUP_PENDING_REASONS: Readonly> = { - 'ownership-fence-lost': true, - 'owner-unavailable': true, - 'transport-failed': true, - 'cleanup-unconfirmed': true, - 'manual-recovery-required': true, -}; - export async function finishLiveDurableCapture< K extends string, H extends LiveResourceHandle, @@ -94,29 +85,28 @@ export async function finishDurableCaptureHandle< // 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: Extract | undefined; + let cleanup: CleanupOutcome; try { - await params.handle[Symbol.asyncDispose](); + cleanup = await params.handle.forceCleanup(); } catch (error) { - const reason = error instanceof AppError ? error.details?.reason : undefined; cleanup = { status: 'cleanup-pending', - reason: - typeof reason === 'string' && Object.hasOwn(CLEANUP_PENDING_REASONS, reason) - ? (reason as CleanupPendingReason) - : 'cleanup-unconfirmed', + 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 (cleanup) throw cleanupPendingError(definition, cleanup); + if (!isConfirmedCleanup(cleanup)) throw cleanupPendingError(definition, cleanup); throw completedDurableCaptureError(definition.store); }, run: async (lease) => { diff --git a/packages/capture-kit/src/durable-resource-cleanup.ts b/packages/capture-kit/src/durable-resource-cleanup.ts new file mode 100644 index 0000000000..0ad129591d --- /dev/null +++ b/packages/capture-kit/src/durable-resource-cleanup.ts @@ -0,0 +1,24 @@ +import { type CleanupOutcome, isConfirmedCleanup } from '@agent-device/contracts/durable-resource'; + +/** 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/capture-kit/src/index.ts b/packages/capture-kit/src/index.ts index 5c83dd4f4e..1fc638a551 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 './durable-resource-cleanup.ts'; 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 bf20e550aa..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,6 +122,9 @@ 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]: dispose, @@ -130,9 +133,7 @@ describe('provider app-log poller', () => { pollerOptions(reader, { writes: [], sleeps, - onOutputDispose: () => { - outputDisposed = true; - }, + onOutputDispose, }), ); await vi.waitFor(() => expect(sleeps.hasPending(1_000)).toBe(true)); @@ -148,6 +149,7 @@ describe('provider app-log poller', () => { 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 9525069782..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,7 +97,7 @@ 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'; @@ -172,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 cd61673b3b..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,5 +1,6 @@ +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'; @@ -7,11 +8,17 @@ 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 Error('Transport unavailable')); + 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 assert.rejects(async () => await handle[Symbol.asyncDispose]()); + 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]()]); diff --git a/packages/capture-kit/src/screen-recording-live-handle.ts b/packages/capture-kit/src/screen-recording-live-handle.ts index e6f830ab8a..d5bdee6413 100644 --- a/packages/capture-kit/src/screen-recording-live-handle.ts +++ b/packages/capture-kit/src/screen-recording-live-handle.ts @@ -1,3 +1,4 @@ +import { createRetryableCleanup } from './durable-resource-cleanup.ts'; import { AppError } from '@agent-device/kernel/errors'; import { type CleanupOutcome, @@ -35,8 +36,6 @@ export function createScreenRecordingLiveHandle( let snapshot = freezeSnapshot(initial); let finish: Promise> | undefined; let terminalRequested = false; - let cleanup: Promise | undefined; - let disposal: Promise | undefined; // Once a finish or a cleanup has been asked for, the recording no longer accepts telemetry. const terminal = (): boolean => terminalRequested; const finishRecording = (progress?: DurableCaptureProgress) => { @@ -59,19 +58,6 @@ export function createScreenRecordingLiveHandle( async () => await implementation.forceCleanup(snapshot), ) : implementation.forceCleanup(snapshot); - const forceCleanup = () => { - terminalRequested = true; - return (cleanup ??= clean().then( - (outcome) => { - if (!isConfirmedCleanup(outcome)) cleanup = undefined; - return outcome; - }, - (error) => { - cleanup = undefined; - throw error; - }, - )); - }; return Object.freeze({ inspect: () => snapshot, appendGestureEvents: (events: readonly RecordingGestureEvent[]) => { @@ -97,16 +83,10 @@ export function createScreenRecordingLiveHandle( snapshot = freezeSnapshot({ ...snapshot, invalidatedReason }); }, finish: finishRecording, - forceCleanup, - [Symbol.asyncDispose]: async () => { - disposal ??= forceCleanup() - .then(assertConfirmedCleanup) - .catch((error) => { - disposal = undefined; - throw error; - }); - await disposal; - }, + ...createRetryableCleanup(() => { + terminalRequested = true; + return clean(); + }, assertConfirmedCleanup), }); } diff --git a/packages/platform-apple/src/logs/start.test.ts b/packages/platform-apple/src/logs/start.test.ts index 463c044881..35f2074a78 100644 --- a/packages/platform-apple/src/logs/start.test.ts +++ b/packages/platform-apple/src/logs/start.test.ts @@ -6,11 +6,13 @@ 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: async () => {}, - [Symbol.asyncDispose]: async () => {}, + terminate, + [Symbol.asyncDispose]: disposeProcess, }), }); const host = { @@ -31,6 +33,8 @@ test('a host log handle retries failed output disposal after its process exited' 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 () => { diff --git a/packages/platform-apple/src/logs/start.ts b/packages/platform-apple/src/logs/start.ts index eb6bc1f966..212cff6682 100644 --- a/packages/platform-apple/src/logs/start.ts +++ b/packages/platform-apple/src/logs/start.ts @@ -278,17 +278,18 @@ 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'; @@ -310,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.ts b/src/platform-runtime-perf-capture-host.ts index 5027ebb181..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,14 +180,8 @@ 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); - const forceCleanup = async () => - (cleanup ??= clean().then((outcome) => { - if (outcome.status === 'cleanup-pending') cleanup = undefined; - return outcome; - })); return Object.freeze({ inspect: steps.inspect, setOutputPath: (value: string) => { @@ -202,11 +197,7 @@ function createCaptureHandle( throw error; }, )), - forceCleanup, - [Symbol.asyncDispose]: async () => { - const outcome = await forceCleanup(); - assertCleanup(outcome); - }, + ...createRetryableCleanup(clean, assertCleanup), }); } From 8c436df63fce0a77dc3ee96773387e8fcdea1da5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Pierzcha=C5=82a?= Date: Mon, 2 Nov 2026 08:10:41 +0100 Subject: [PATCH 9/9] refactor(capture): colocate cleanup retries with the durable contract --- .../capture-kit/src/app-log-live-handle.ts | 2 +- .../capture-kit/src/audio-probe-runtime.ts | 2 +- .../src/durable-resource-cleanup.ts | 24 ------------------- packages/capture-kit/src/index.ts | 2 +- .../src/screen-recording-live-handle.ts | 2 +- packages/contracts/src/durable-resource.ts | 23 ++++++++++++++++++ 6 files changed, 27 insertions(+), 28 deletions(-) delete mode 100644 packages/capture-kit/src/durable-resource-cleanup.ts diff --git a/packages/capture-kit/src/app-log-live-handle.ts b/packages/capture-kit/src/app-log-live-handle.ts index 1963f76aa8..a2c4d9c35d 100644 --- a/packages/capture-kit/src/app-log-live-handle.ts +++ b/packages/capture-kit/src/app-log-live-handle.ts @@ -1,4 +1,3 @@ -import { createRetryableCleanup } from './durable-resource-cleanup.ts'; import { AppError } from '@agent-device/kernel/errors'; import type { AppLogCompletion, @@ -9,6 +8,7 @@ import { type CleanupOutcome, type FinishOutcome, isConfirmedCleanup, + createRetryableCleanup, } from '@agent-device/contracts/durable-resource'; type AppLogLiveHandleImplementation = Readonly<{ diff --git a/packages/capture-kit/src/audio-probe-runtime.ts b/packages/capture-kit/src/audio-probe-runtime.ts index f1da23b94e..6ec6353313 100644 --- a/packages/capture-kit/src/audio-probe-runtime.ts +++ b/packages/capture-kit/src/audio-probe-runtime.ts @@ -1,4 +1,3 @@ -import { createRetryableCleanup } from './durable-resource-cleanup.ts'; import fs from 'node:fs/promises'; import { deviceIdentity, type DeviceInfo } from '@agent-device/kernel/device'; import { AppError } from '@agent-device/kernel/errors'; @@ -17,6 +16,7 @@ import type { } from '@agent-device/contracts/audio-probe-runtime-host'; import { isConfirmedCleanup, + createRetryableCleanup, type CleanupOutcome, type FinishOutcome, type ReattachOutcome, diff --git a/packages/capture-kit/src/durable-resource-cleanup.ts b/packages/capture-kit/src/durable-resource-cleanup.ts deleted file mode 100644 index 0ad129591d..0000000000 --- a/packages/capture-kit/src/durable-resource-cleanup.ts +++ /dev/null @@ -1,24 +0,0 @@ -import { type CleanupOutcome, isConfirmedCleanup } from '@agent-device/contracts/durable-resource'; - -/** 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/capture-kit/src/index.ts b/packages/capture-kit/src/index.ts index 1fc638a551..afc77befd0 100644 --- a/packages/capture-kit/src/index.ts +++ b/packages/capture-kit/src/index.ts @@ -10,7 +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 './durable-resource-cleanup.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/screen-recording-live-handle.ts b/packages/capture-kit/src/screen-recording-live-handle.ts index d5bdee6413..14ce27af82 100644 --- a/packages/capture-kit/src/screen-recording-live-handle.ts +++ b/packages/capture-kit/src/screen-recording-live-handle.ts @@ -1,10 +1,10 @@ -import { createRetryableCleanup } from './durable-resource-cleanup.ts'; import { AppError } from '@agent-device/kernel/errors'; import { type CleanupOutcome, type DurableCaptureProgress, type FinishOutcome, isConfirmedCleanup, + createRetryableCleanup, } from '@agent-device/contracts/durable-resource'; import type { RecordingGestureEvent, 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()), + }; +}