diff --git a/CONTEXT.md b/CONTEXT.md index 4242313a3..04c1a3fb6 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -91,7 +91,13 @@ task touches: (`active`/`expired`), and an issuance scope (`all` for a complete snapshot, or the bounded set of ref bodies a partial publication emitted). Owned solely by `src/daemon/ref-frame.ts`. A complete snapshot activates an `all` frame; `find`/settled diff/replay divergence activate a bounded partial - frame that supersedes the prior one; internal read captures never activate or reindex it. + frame that supersedes the prior one; internal read captures never activate or reindex it. Replay + captures return opaque, one-shot lineage evidence, and daemon response composition activates refs + synchronously only after the exact inline or successfully written overflow projection is known. + Every finalization attempt consumes its evidence. The outer replay retains its stable session lock + plus the device lock when known through finalization, so external commands cannot interleave; + nested replay actions reuse that scope and invalidate lineage through capture, ref-frame, + side-effect, or session-lifetime changes. - Frame expiry seam (ADR 0014): every mutating leaf calls `expireRefFrame` synchronously, immediately before the device op that may change element identity (after all pre-action guards), so a post-dispatch failure still leaves the frame expired — there is no success-only rollback. Ref diff --git a/docs/daemon-modularity-proposal.md b/docs/daemon-modularity-proposal.md index 9ea504073..5b22269a0 100644 --- a/docs/daemon-modularity-proposal.md +++ b/docs/daemon-modularity-proposal.md @@ -297,8 +297,9 @@ export interface AdReplayRuntime { executeStep(input: AdStepExecution): Promise; /** - * Capture a replay observation and atomically update the session observation plus - * the exact partial ref frame exposed by the outcome. + * Capture replay evidence and update operational observation state only. + * The returned value may carry opaque, one-shot lineage evidence; it cannot + * publish or replace client ref authority. */ observeReplay(input: AdObservationRequest): Promise; @@ -323,8 +324,15 @@ export type AdStepExecution = { The daemon runtime adapter owns translation to `DaemonRequest`, parent flag merging, `internal.replayPlanStep`, target/landmark guards, session-scope inheritance through `internal.resolvedSessionScope`, request-level provider-scope reuse, and response/error projection. -`observeReplay` deliberately combines capture, stored observation update, and exact partial-frame -activation. Splitting those operations would expose an illegal intermediate state under ADR 0014. +`observeReplay` updates the stored operational observation and returns opaque, one-shot +capture-lineage evidence. Daemon response composition projects the exact inline response or +successfully written overflow artifact, consumes and validates that evidence synchronously, and only +then activates the matching partial frame. Every finalization attempt consumes the token, including +empty, cancelled, invalid, and stale outcomes. The engine cannot publish refs or access session/store +authority; no asynchronous work may occur between activation and returning the projected response. +The outer replay keeps its stable session lock plus the device lock when known through this +finalization. Same-session nested actions reuse the admitted scope; their meaningful changes +invalidate lineage through observation, ref-frame, runtime-revision, or session-lifetime checks. The engine receives no session interface. A daemon-owned coordinator wraps `execute` inside the already locked scope: @@ -854,8 +862,8 @@ This is mostly ownership correction around an already-proven port. - Keep canonical `.ad` syntax in a pure internal codec shared with session publication. Publication keeps portability/destination validation, atomic commit, and lifecycle authority. - Preserve inherited session scope, request-level provider scope, the existing lock, and the - side-effect/ref-frame seam. `observeReplay` atomically couples capture and partial-frame - activation. + side-effect/ref-frame seam. `observeReplay` updates operational observation and returns opaque + lineage evidence; daemon response composition alone activates the exact projected partial frame. - Delete superseded helpers and handler-mock tests as their interface replacements land. ### Phase 5: narrow platform binding diff --git a/src/compat/maestro/__tests__/daemon-runtime-port-support.test.ts b/src/compat/maestro/__tests__/daemon-runtime-port-support.test.ts index 020743b9f..30dba2202 100644 --- a/src/compat/maestro/__tests__/daemon-runtime-port-support.test.ts +++ b/src/compat/maestro/__tests__/daemon-runtime-port-support.test.ts @@ -38,6 +38,27 @@ test('composes operation-specific Maestro flags with the runtime envelope', asyn ); }); +test('marks Maestro hierarchy captures as daemon-private observations', async () => { + const invoke = vi.fn(async () => ({ ok: true as const, data: { nodes: [] } })); + + await invokeMaestroPublicOperation( + { + baseReq: makeBaseRequest(), + invoke, + dependencies: makeDependencies(), + platform: 'ios', + }, + { kind: 'snapshot' }, + ); + + expect(invoke).toHaveBeenCalledWith( + expect.objectContaining({ + command: 'snapshot', + internal: expect.objectContaining({ observationOnly: true }), + }), + ); +}); + test('preserves diagnostic metadata carried inside daemon error details', async () => { const invoke = vi.fn(async () => ({ ok: false as const, diff --git a/src/compat/maestro/daemon-runtime-port-support.ts b/src/compat/maestro/daemon-runtime-port-support.ts index e1c092f0b..e2389082f 100644 --- a/src/compat/maestro/daemon-runtime-port-support.ts +++ b/src/compat/maestro/daemon-runtime-port-support.ts @@ -51,8 +51,11 @@ export async function invokeMaestroPublicOperation( ...baseReq } = options.baseReq; const effectiveFlags = flagsWith(baseFlags, projected.flags ?? {}); - const effectiveInternal = - projected.internal === undefined ? baseInternal : { ...baseInternal, ...projected.internal }; + const effectiveInternal = stripUndefined({ + ...baseInternal, + ...projected.internal, + ...(operation.kind === 'snapshot' ? { observationOnly: true as const } : {}), + }); const response = await options.invoke( stripUndefined({ ...baseReq, @@ -60,7 +63,7 @@ export async function invokeMaestroPublicOperation( positionals: projected.positionals, input: projected.input, flags: effectiveFlags, - internal: effectiveInternal, + internal: Object.keys(effectiveInternal).length > 0 ? effectiveInternal : undefined, }), ); if (!response.ok) throw daemonResponseError(response); diff --git a/src/daemon/__tests__/http-server-rpc-validation.test.ts b/src/daemon/__tests__/http-server-rpc-validation.test.ts index 73c1b39a7..66c226f72 100644 --- a/src/daemon/__tests__/http-server-rpc-validation.test.ts +++ b/src/daemon/__tests__/http-server-rpc-validation.test.ts @@ -78,12 +78,12 @@ test('malformed command params (command as number) yield 400 / -32602', async (t }); // `DaemonRequest.internal` carries semantics-affecting bits stamped inside the -// daemon — `replayPlanStep` (#1271 stage 2) decides whether a read is an -// authored plan step or an out-of-band diagnostic, and so whether it lands in a -// repair heal. Two independent allowlists keep it unreachable from the wire: -// `commandRpcParamsSchema` projects only its eight named fields, and +// daemon — `replayPlanStep` controls recording provenance and +// `observationOnly` suppresses snapshot ref issuance for daemon-composed +// Maestro reads. Two independent allowlists keep both unreachable from the +// HTTP wire: `commandRpcParamsSchema` projects only its eight named fields, and // `toDaemonRequest` then builds the request field by field. Both would have to -// regress for a caller to stamp its own provenance; this pins the resulting +// regress for a caller to stamp these semantics; this pins the resulting // boundary contract so neither drifts silently. test('the rpc boundary never accepts internal request fields from the wire', async (t) => { if (await skipWhenLoopbackUnavailable(t)) return; @@ -105,9 +105,13 @@ test('the rpc boundary never accepts internal request fields from the wire', asy id: 'req-1', method: 'agent_device.command', params: { - command: 'get', - positionals: ['text', 'id=whatever'], - internal: { replayPlanStep: true, replayTargetGuard: { ref: '@e1' } }, + command: 'snapshot', + positionals: [], + internal: { + observationOnly: true, + replayPlanStep: true, + replayTargetGuard: { ref: '@e1' }, + }, }, }), }); diff --git a/src/daemon/__tests__/internal-observation.test.ts b/src/daemon/__tests__/internal-observation.test.ts new file mode 100644 index 000000000..e57931726 --- /dev/null +++ b/src/daemon/__tests__/internal-observation.test.ts @@ -0,0 +1,197 @@ +import os from 'node:os'; +import path from 'node:path'; +import { expect, test } from 'vitest'; +import { makeIosSession } from '../../__tests__/test-utils/session-factories.ts'; +import type { SnapshotState } from '../../kernel/snapshot.ts'; +import { bindInternalObservationAuthority } from '../internal-observation.ts'; +import { expireRefFrame } from '../ref-frame.ts'; +import { markSessionPartialRefsIssued, setSessionSnapshot } from '../session-snapshot.ts'; +import { SessionStore } from '../session-store.ts'; + +function snapshot(ref: string, label = ref): SnapshotState { + return { + createdAt: Date.now(), + nodes: [ + { + index: 0, + depth: 0, + type: 'Button', + ref, + label, + rect: { x: 0, y: 0, width: 100, height: 44 }, + hittable: true, + }, + ], + }; +} + +function scenario() { + const root = path.join(os.tmpdir(), `agent-device-internal-observation-${crypto.randomUUID()}`); + const sessionStore = new SessionStore(path.join(root, 'sessions')); + const sessionName = 'default'; + const session = makeIosSession(sessionName, { appBundleId: 'com.example.app' }); + const prior = snapshot('e1', 'Previously published'); + setSessionSnapshot(session, prior); + markSessionPartialRefsIssued(session, ['e1']); + sessionStore.set(sessionName, session); + const captured = snapshot('e2', 'Internal capture'); + const authority = bindInternalObservationAuthority({ + sessionStore, + sessionName, + }); + const stored = authority.store(captured); + return { sessionStore, sessionName, session, prior, captured, authority, ...stored }; +} + +function publishCurrent(input: ReturnType, signal?: AbortSignal) { + const authority = bindInternalObservationAuthority({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + ...(signal ? { signal } : {}), + }); + return authority.finalize(input.evidence, { + refsGeneration: input.refsGeneration, + refs: ['@e2'], + }); +} + +test('publishes exactly the validated outward refs from a current internal capture', () => { + const input = scenario(); + + expect(publishCurrent(input)).toEqual({ + published: true, + refsGeneration: input.refsGeneration, + refCount: 1, + }); + expect(input.session.refFrameState).toBe('active'); + expect(input.session.refFrameScope).toEqual(new Set(['e2'])); + expect(input.session.refFrameTree).toBe(input.captured); + expect(input.session.refFrameGeneration).toBe(input.refsGeneration); + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); +}); + +test('empty publication never supersedes prior client authority', () => { + const input = scenario(); + + const result = input.authority.finalize(input.evidence, { + refsGeneration: input.refsGeneration, + refs: [], + }); + + expect(result).toEqual({ published: false, reason: 'empty' }); + expect(input.session.refFrameScope).toEqual(new Set(['e1'])); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('an empty finalization consumes its evidence and cannot later publish', () => { + const input = scenario(); + + const first = input.authority.finalize(input.evidence, { + refsGeneration: input.refsGeneration, + refs: [], + }); + + expect(first).toEqual({ published: false, reason: 'empty' }); + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + expect(input.session.refFrameScope).toEqual(new Set(['e1'])); +}); + +test('cancelled publication leaves prior client authority intact', () => { + const input = scenario(); + const controller = new AbortController(); + controller.abort(); + + expect(publishCurrent(input, controller.signal)).toEqual({ + published: false, + reason: 'cancelled', + }); + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + expect(input.session.refFrameScope).toEqual(new Set(['e1'])); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('a newer capture makes older capture evidence stale', () => { + const input = scenario(); + setSessionSnapshot(input.session, snapshot('e3', 'Later capture')); + + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + expect(input.session.refFrameScope).toEqual(new Set(['e1'])); +}); + +test('a later ref publication prevents an older capture from superseding it', () => { + const input = scenario(); + markSessionPartialRefsIssued(input.session, ['e2']); + const laterTree = input.session.refFrameTree; + + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + expect(input.session.refFrameTree).toBe(laterTree); +}); + +test('a runtime side effect invalidates capture evidence without rolling authority back', () => { + const input = scenario(); + expireRefFrame(input.session); + + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + expect(input.session.refFrameState).toBe('expired'); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('runtime revision invalidates evidence even when the ref frame was already expired', () => { + const input = scenario(); + expireRefFrame(input.session); + const recaptured = input.authority.store(snapshot('e3', 'Captured after expiry')); + expireRefFrame(input.session); + + const result = input.authority.finalize(recaptured.evidence, { + refsGeneration: recaptured.refsGeneration, + refs: ['e3'], + }); + + expect(result).toEqual({ published: false, reason: 'stale-capture' }); + expect(input.session.refFrameState).toBe('expired'); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('session close invalidates capture evidence', () => { + const input = scenario(); + input.sessionStore.delete(input.sessionName); + + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + // Even restoring the exact same session object cannot revive evidence that + // a stale finalization attempt already consumed. + input.sessionStore.set(input.sessionName, input.session); + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + expect(input.session.refFrameScope).toEqual(new Set(['e1'])); +}); + +test('same-name session replacement cannot inherit capture evidence', () => { + const input = scenario(); + const replacement = makeIosSession(input.sessionName, { appBundleId: 'com.example.app' }); + input.sessionStore.set(input.sessionName, replacement); + + expect(publishCurrent(input)).toEqual({ published: false, reason: 'stale-capture' }); + expect(replacement.refFrameTree).toBeUndefined(); +}); + +test('generation and ref projection must match the exact captured tree', () => { + const wrongGeneration = scenario(); + const generationResult = wrongGeneration.authority.finalize(wrongGeneration.evidence, { + refsGeneration: wrongGeneration.refsGeneration + 1, + refs: ['e2'], + }); + expect(generationResult).toEqual({ published: false, reason: 'invalid-projection' }); + expect(publishCurrent(wrongGeneration)).toEqual({ + published: false, + reason: 'stale-capture', + }); + + const wrongRef = scenario(); + const refResult = wrongRef.authority.finalize(wrongRef.evidence, { + refsGeneration: wrongRef.refsGeneration, + refs: ['e999'], + }); + expect(refResult).toEqual({ published: false, reason: 'invalid-projection' }); + expect(publishCurrent(wrongRef)).toEqual({ published: false, reason: 'stale-capture' }); + expect(wrongGeneration.session.refFrameScope).toEqual(new Set(['e1'])); + expect(wrongRef.session.refFrameScope).toEqual(new Set(['e1'])); +}); diff --git a/src/daemon/__tests__/request-execution-locks.test.ts b/src/daemon/__tests__/request-execution-locks.test.ts new file mode 100644 index 000000000..dd4b13c01 --- /dev/null +++ b/src/daemon/__tests__/request-execution-locks.test.ts @@ -0,0 +1,55 @@ +import { expect, test } from 'vitest'; +import { createRequestExecutionLocks } from '../request-execution-locks.ts'; + +test('a dynamically retained device lock lasts until the owning request completes', async () => { + const locks = new Map>(); + const replayLocks = createRequestExecutionLocks({ + locks, + initialKeys: ['session:replay'], + }); + const externalLocks = createRequestExecutionLocks({ + locks, + initialKeys: ['session:external', 'device:SIM-001'], + }); + let finishReplay: () => void = () => {}; + const replayFinished = new Promise((resolve) => { + finishReplay = resolve; + }); + let markDeviceRetained: () => void = () => {}; + const deviceRetained = new Promise((resolve) => { + markDeviceRetained = resolve; + }); + + const replayRun = replayLocks.run(async () => { + await replayLocks.retainDevice('SIM-001'); + markDeviceRetained(); + await replayFinished; + }); + await deviceRetained; + + let externalEntered = false; + const externalRun = externalLocks.run(async () => { + externalEntered = true; + }); + await Promise.resolve(); + expect(externalEntered).toBe(false); + + finishReplay(); + await replayRun; + await externalRun; + expect(externalEntered).toBe(true); +}); + +test('retaining an initial device lock is reentrant within the request', async () => { + const requestLocks = createRequestExecutionLocks({ + locks: new Map(), + initialKeys: ['session:replay', 'device:SIM-001'], + }); + + await expect( + requestLocks.run(async () => { + await requestLocks.retainDevice('SIM-001'); + return 'done'; + }), + ).resolves.toBe('done'); +}); diff --git a/src/daemon/__tests__/request-execution-scope.test.ts b/src/daemon/__tests__/request-execution-scope.test.ts index c040a4b38..3d1d065fb 100644 --- a/src/daemon/__tests__/request-execution-scope.test.ts +++ b/src/daemon/__tests__/request-execution-scope.test.ts @@ -241,6 +241,90 @@ test('leased session heartbeat is serialized with the request execution lock', a expect(leaseRegistry.listActiveLeases()[0]?.heartbeatAt).toBe(3_000); }); +test('a later external command cannot interleave with replay observation finalization', async () => { + const sessionStore = makeSessionStore('agent-device-request-scope-'); + sessionStore.set('default', makeIosSession('default')); + const leaseRegistry = new LeaseRegistry(); + const replay = await createRequestExecutionScope({ + req: makeRequest({ command: 'replay' }), + sessionStore, + leaseRegistry, + }); + const laterSnapshot = await createRequestExecutionScope({ + req: makeRequest({ command: 'snapshot' }), + sessionStore, + leaseRegistry, + }); + + let finishReplay: () => void = () => {}; + let replayEntered: () => void = () => {}; + const replayEnteredPromise = new Promise((resolve) => { + replayEntered = resolve; + }); + const replayRun = replay.runLocked( + async () => + await new Promise((resolve) => { + finishReplay = resolve; + replayEntered(); + }), + ); + await replayEnteredPromise; + + let laterCommandEntered = false; + const laterRun = laterSnapshot.runLocked(async () => { + laterCommandEntered = true; + }); + await Promise.resolve(); + expect(laterCommandEntered).toBe(false); + + finishReplay(); + await replayRun; + await laterRun; + expect(laterCommandEntered).toBe(true); +}); + +test('a fresh replay keeps its session lock after a nested open binds the device', async () => { + const sessionStore = makeSessionStore('agent-device-request-scope-'); + const leaseRegistry = new LeaseRegistry(); + const replay = await createRequestExecutionScope({ + req: makeRequest({ command: 'replay' }), + sessionStore, + leaseRegistry, + }); + + let finishReplay: () => void = () => {}; + let sessionOpened: () => void = () => {}; + const sessionOpenedPromise = new Promise((resolve) => { + sessionOpened = resolve; + }); + const replayRun = replay.runLocked( + async () => + await new Promise((resolve) => { + sessionStore.set('default', makeIosSession('default')); + finishReplay = resolve; + sessionOpened(); + }), + ); + await sessionOpenedPromise; + + const laterSnapshot = await createRequestExecutionScope({ + req: makeRequest({ command: 'snapshot' }), + sessionStore, + leaseRegistry, + }); + let laterCommandEntered = false; + const laterRun = laterSnapshot.runLocked(async () => { + laterCommandEntered = true; + }); + await Promise.resolve(); + expect(laterCommandEntered).toBe(false); + + finishReplay(); + await replayRun; + await laterRun; + expect(laterCommandEntered).toBe(true); +}); + test('leased session rejects mismatched lease id before dispatch', async () => { const sessionStore = makeSessionStore('agent-device-request-scope-'); const leaseRegistry = new LeaseRegistry(); diff --git a/src/daemon/__tests__/request-router-replay-scope.test.ts b/src/daemon/__tests__/request-router-replay-scope.test.ts index de86e37ce..82d044b23 100644 --- a/src/daemon/__tests__/request-router-replay-scope.test.ts +++ b/src/daemon/__tests__/request-router-replay-scope.test.ts @@ -33,15 +33,19 @@ import { makeIosSession } from '../../__tests__/test-utils/session-factories.ts' import { makeSessionStore } from '../../__tests__/test-utils/store-factory.ts'; import { LeaseRegistry } from '../lease-registry.ts'; import { createRequestHandler } from '../request-router.ts'; +import { ensureDeviceReady } from '../device-ready.ts'; const mockDispatch = vi.mocked(dispatchCommand); const mockResolveTargetDevice = vi.mocked(getResolveTargetDeviceMock()); +const mockEnsureDeviceReady = vi.mocked(ensureDeviceReady); beforeEach(() => { mockDispatch.mockReset(); mockDispatch.mockResolvedValue({}); mockResolveTargetDevice.mockReset(); mockResolveTargetDevice.mockResolvedValue(IOS_SIMULATOR); + mockEnsureDeviceReady.mockReset(); + mockEnsureDeviceReady.mockResolvedValue(); }); test('replay runs active-session actions inside the parent request provider scope', async () => { @@ -125,7 +129,7 @@ test('session list includes a cwd-scoped session opened by replay', async () => command: 'replay', positionals: [replayPath], flags: { platform: 'ios' }, - meta: { cwd: root, requestId: 'replay-open-scope' }, + meta: { cwd: root, requestId: 'replay-open-scope', deviceKey: 'test:sim-1' }, }); expect(replayResponse).toMatchObject({ ok: true }); @@ -148,3 +152,67 @@ test('session list includes a cwd-scoped session opened by replay', async () => }, }); }); + +test('fresh replay retains a dynamically selected device through finalization', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'agent-device-replay-device-lock-')); + const replayPath = path.join(root, 'flow.ad'); + fs.writeFileSync( + replayPath, + 'runtime set --platform ios --metro-host localhost\nopen "demo://checkout"\n', + ); + const sessionStore = makeSessionStore('agent-device-replay-device-lock-'); + const leaseRegistry = new LeaseRegistry(); + let releaseReadiness: () => void = () => {}; + const readinessReleased = new Promise((resolve) => { + releaseReadiness = resolve; + }); + let markReadinessEntered: () => void = () => {}; + const readinessEntered = new Promise((resolve) => { + markReadinessEntered = resolve; + }); + mockEnsureDeviceReady.mockImplementation(async () => { + markReadinessEntered(); + await readinessReleased; + }); + + const handler = createRequestHandler({ + logPath: path.join(os.tmpdir(), 'daemon.log'), + token: 'test-token', + sessionStore, + leaseRegistry, + appleRunnerProvider: () => undefined, + trackDownloadableArtifact: () => 'artifact-id', + }); + const replayResponse = handler({ + token: 'test-token', + session: 'default', + command: 'replay', + positionals: [replayPath], + flags: { platform: 'ios' }, + meta: { cwd: root, requestId: 'replay-device-lock', deviceKey: 'test:sim-1' }, + }); + await readinessEntered; + + sessionStore.set('external', makeIosSession('external')); + let externalRequestSettled = false; + const externalResponse = handler({ + token: 'test-token', + session: 'external', + command: 'snapshot', + positionals: [], + meta: { + cwd: root, + requestId: 'external-device-command', + sessionExplicit: true, + }, + }).finally(() => { + externalRequestSettled = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(externalRequestSettled).toBe(false); + + releaseReadiness(); + await expect(replayResponse).resolves.toMatchObject({ ok: true }); + await externalResponse; + expect(externalRequestSettled).toBe(true); +}); diff --git a/src/daemon/handlers/__tests__/session-replay-divergence-publication.test.ts b/src/daemon/handlers/__tests__/session-replay-divergence-publication.test.ts new file mode 100644 index 000000000..39401a3de --- /dev/null +++ b/src/daemon/handlers/__tests__/session-replay-divergence-publication.test.ts @@ -0,0 +1,325 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { beforeEach, expect, test, vi } from 'vitest'; + +vi.mock('../../../core/dispatch.ts', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, dispatchCommand: vi.fn(async () => ({})) }; +}); + +import { dispatchCommand } from '../../../core/dispatch.ts'; +import { makeIosSession } from '../../../__tests__/test-utils/session-factories.ts'; +import type { SnapshotState } from '../../../kernel/snapshot.ts'; +import type { ReplayDivergence } from '../../../replay/divergence.ts'; +import { bindInternalObservationAuthority } from '../../internal-observation.ts'; +import { expireRefFrame } from '../../ref-frame.ts'; +import { markSessionPartialRefsIssued, setSessionSnapshot } from '../../session-snapshot.ts'; +import { SessionStore } from '../../session-store.ts'; +import { captureDivergenceObservation } from '../session-replay-divergence.ts'; +import { boundReplayDivergenceForSession } from '../session-replay-divergence-publication.ts'; + +const mockDispatchCommand = vi.mocked(dispatchCommand); + +beforeEach(() => { + mockDispatchCommand.mockReset(); + mockDispatchCommand.mockResolvedValue({}); +}); + +function scenario(refCount = 20) { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'agent-device-divergence-publication-')); + const sessionStore = new SessionStore(path.join(root, 'sessions')); + const sessionName = 'default'; + const session = makeIosSession(sessionName, { appBundleId: 'com.example.app' }); + const prior: SnapshotState = { + createdAt: 1, + nodes: [{ index: 0, depth: 0, type: 'Button', ref: 'old', label: 'Old' }], + }; + setSessionSnapshot(session, prior); + markSessionPartialRefsIssued(session, ['old']); + sessionStore.set(sessionName, session); + const snapshot: SnapshotState = { + createdAt: 2, + nodes: Array.from({ length: refCount }, (_, index) => ({ + index, + depth: 0, + type: 'Button', + ref: `e${index + 1}`, + label: `Button ${index + 1}`, + rect: { x: 0, y: index * 44, width: 100, height: 44 }, + hittable: true, + })), + }; + const authority = bindInternalObservationAuthority({ + sessionStore, + sessionName, + }); + const observation = authority.store(snapshot); + return { root, sessionStore, sessionName, session, prior, snapshot, ...observation }; +} + +function divergence( + input: ReturnType, + overrides: Partial = {}, +): ReplayDivergence { + return { + version: 1, + kind: 'action-failure', + step: { index: 1, source: { path: '/tmp/flow.ad', line: 1 } }, + action: 'press label="Save"', + cause: { code: 'COMMAND_FAILED', message: 'failed' }, + screen: { + state: 'available', + refsGeneration: input.refsGeneration, + refs: input.snapshot.nodes.map((node) => ({ + ref: node.ref!, + role: 'button', + label: node.label, + })), + }, + suggestions: [], + suggestionCount: 0, + resume: { allowed: false, from: 1, planDigest: 'digest', reason: 'failed' }, + repairHint: 'manual', + ...overrides, + }; +} + +test('internal divergence capture updates observation without publishing client ref authority', async () => { + const input = scenario(1); + mockDispatchCommand.mockResolvedValue({ + nodes: [ + { + index: 0, + depth: 0, + type: 'Button', + label: 'Captured again', + rect: { x: 0, y: 50, width: 100, height: 44 }, + hittable: true, + }, + ], + truncated: false, + backend: 'xctest', + }); + + const observation = await captureDivergenceObservation({ + session: input.session, + sessionName: input.sessionName, + sessionStore: input.sessionStore, + logPath: path.join(input.root, 'daemon.log'), + action: { + command: 'press', + positionals: ['label="Captured again"'], + flags: {}, + result: {}, + }, + }); + + expect(observation.state).toBe('available'); + expect(input.session.snapshot?.nodes[0]?.label).toBe('Captured again'); + expect(input.session.refFrameScope).toEqual(new Set(['old'])); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test.each([ + ['digest', 8], + ['default', 20], + ['full', 20], +] as const)('publishes exactly the direct %s response-level ref cap', (level, expected) => { + const input = scenario(); + + const result = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input), + responseLevel: level, + evidence: input.evidence, + }); + + expect(result.screen.state).toBe('available'); + const refs = + result.screen.state === 'available' ? result.screen.refs.map((entry) => entry.ref) : []; + expect(refs).toHaveLength(expected); + expect(input.session.refFrameScope).toEqual(new Set(refs)); + expect(input.session.refFrameTree).toBe(input.snapshot); +}); + +test('missing capture evidence fails closed instead of exposing unauthorized refs', () => { + const input = scenario(); + + const result = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input), + responseLevel: 'default', + evidence: undefined, + }); + + expect(result.screen).toEqual( + expect.objectContaining({ + state: 'unavailable', + reason: 'ref-publication-missing-evidence', + }), + ); + expect(input.session.refFrameScope).toEqual(new Set(['old'])); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('degenerate projected refs fail closed when publication normalizes to empty', () => { + const input = scenario(); + const result = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input, { + screen: { + state: 'available', + refsGeneration: input.refsGeneration, + refs: [ + { ref: '@', role: 'button', label: 'Empty ref' }, + { ref: '@~s3', role: 'button', label: 'Empty scoped ref' }, + ], + }, + suggestions: [ + { selector: 'label="Empty ref"', basis: 'label', ref: '@' }, + { selector: 'label="Empty scoped ref"', basis: 'label', ref: '@~s3' }, + ], + suggestionCount: 2, + }), + responseLevel: 'default', + evidence: input.evidence, + }); + + expect(result.screen).toEqual( + expect.objectContaining({ + state: 'unavailable', + reason: 'ref-publication-empty', + }), + ); + expect(result.suggestions).toEqual([ + { selector: 'label="Empty ref"', basis: 'label' }, + { selector: 'label="Empty scoped ref"', basis: 'label' }, + ]); + expect(input.session.refFrameScope).toEqual(new Set(['old'])); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('an empty outward projection consumes its one-shot capture evidence', () => { + const input = scenario(); + const empty = divergence(input, { + screen: { + state: 'available', + refsGeneration: input.refsGeneration, + refs: [], + }, + }); + + const first = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: empty, + responseLevel: 'default', + evidence: input.evidence, + }); + const second = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input), + responseLevel: 'default', + evidence: input.evidence, + }); + + expect(first.screen).toEqual(empty.screen); + expect(second.screen).toEqual( + expect.objectContaining({ + state: 'unavailable', + reason: 'ref-publication-stale-capture', + }), + ); + expect(input.session.refFrameScope).toEqual(new Set(['old'])); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('successful overflow publishes the refs exposed by the exact artifact projection', () => { + const input = scenario(); + const result = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input, { + cause: { code: 'COMMAND_FAILED', message: 'x'.repeat(40_000) }, + }), + responseLevel: 'default', + evidence: input.evidence, + }); + + expect(result.screen.state).toBe('unavailable'); + expect(result.overflow?.artifactPath).toBeTruthy(); + const artifact = JSON.parse(fs.readFileSync(result.overflow!.artifactPath, 'utf8')) as { + screen: { refs: Array<{ ref: string }> }; + }; + const artifactRefs = artifact.screen.refs.map((entry) => entry.ref); + expect(input.session.refFrameScope).toEqual(new Set(artifactRefs)); + expect(input.session.refFrameTree).toBe(input.snapshot); +}); + +test('failed overflow artifact with no inline refs publishes nothing', () => { + const input = scenario(); + const blockedArtifactDir = path.join( + input.sessionStore.ensureSessionDir(input.sessionName), + 'replay-divergence', + ); + fs.writeFileSync(blockedArtifactDir, 'not a directory'); + + const result = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input, { + cause: { code: 'COMMAND_FAILED', message: 'x'.repeat(40_000) }, + }), + responseLevel: 'default', + evidence: input.evidence, + }); + + expect(result.artifactUnavailable).toBe(true); + expect(result.screen.state).toBe('unavailable'); + expect(input.session.refFrameScope).toEqual(new Set(['old'])); + expect(input.session.refFrameTree).toBe(input.prior); +}); + +test('stale capture suppresses outward refs and preserves newer authority', () => { + const input = scenario(); + setSessionSnapshot(input.session, { + createdAt: 3, + nodes: [{ index: 0, depth: 0, type: 'Button', ref: 'newer', label: 'Newer' }], + }); + + const result = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input), + responseLevel: 'default', + evidence: input.evidence, + }); + + expect(result.screen.state).toBe('unavailable'); + expect(input.session.refFrameScope).toEqual(new Set(['old'])); +}); + +test('cancellation after capture suppresses outward refs without reactivating an expired frame', () => { + const input = scenario(); + expireRefFrame(input.session); + const controller = new AbortController(); + controller.abort(); + + const result = boundReplayDivergenceForSession({ + sessionStore: input.sessionStore, + sessionName: input.sessionName, + divergence: divergence(input), + responseLevel: 'default', + evidence: input.evidence, + signal: controller.signal, + }); + + expect(result.screen.state).toBe('unavailable'); + expect(input.session.refFrameState).toBe('expired'); + expect(input.session.refFrameTree).toBe(input.prior); +}); diff --git a/src/daemon/handlers/__tests__/session-replay-maestro-failure.test.ts b/src/daemon/handlers/__tests__/session-replay-maestro-failure.test.ts index dea8d4a33..912b447bd 100644 --- a/src/daemon/handlers/__tests__/session-replay-maestro-failure.test.ts +++ b/src/daemon/handlers/__tests__/session-replay-maestro-failure.test.ts @@ -42,10 +42,14 @@ function makeMaestroPlan(): MaestroReplayPlan { }; } -async function buildFailureResponse( +async function buildFailureScenario( command: MaestroCommand, nodes: SnapshotNode[], -): Promise>, { ok: false }>> { +): Promise<{ + response: Extract>, { ok: false }>; + sessionStore: SessionStore; + sessionName: string; +}> { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'agent-device-maestro-suggestions-')); const sessionName = 'default'; const sessionStore = new SessionStore(path.join(root, 'sessions')); @@ -71,7 +75,14 @@ async function buildFailureResponse( logPath: path.join(root, 'daemon.log'), }); if (response.ok) throw new Error('expected typed Maestro failure response'); - return response; + return { response, sessionStore, sessionName }; +} + +async function buildFailureResponse( + command: MaestroCommand, + nodes: SnapshotNode[], +): Promise>, { ok: false }>> { + return (await buildFailureScenario(command, nodes)).response; } test('typed Maestro failure projection keeps the event command and source provenance', () => { @@ -305,6 +316,41 @@ test('typed Maestro failure diagnostics never render inputText payloads', async expect(response.error.message).toContain('inputText'); }); +test('typed Maestro failure publishes exactly the refs exposed by its divergence', async () => { + const command = { + kind: 'tapOn' as const, + source: { path: '/flows/actions.yaml', line: 4 }, + target: { space: 'target' as const, selector: { label: 'Missing' } }, + } satisfies Extract; + const scenario = await buildFailureScenario(command, [ + { + ref: 'e1', + index: 0, + type: 'Application', + rect: { x: 0, y: 0, width: 402, height: 874 }, + }, + { + ref: 'e2', + index: 1, + parentIndex: 0, + label: 'Available action', + type: 'Button', + rect: { x: 16, y: 40, width: 140, height: 44 }, + hittable: true, + }, + ]); + const divergence = scenario.response.error.details?.divergence as { + screen: { state: string; refs: Array<{ ref: string }> }; + }; + const exposedRefs = divergence.screen.refs.map(({ ref }) => ref); + + expect(divergence.screen.state).toBe('available'); + expect(exposedRefs).toEqual(['e2']); + expect(scenario.sessionStore.get(scenario.sessionName)?.refFrameScope).toEqual( + new Set(exposedRefs), + ); +}); + test('typed Maestro suggestions rank visible childOf candidates and exclude out-of-scope nodes', async () => { const command = { kind: 'tapOn' as const, diff --git a/src/daemon/handlers/__tests__/snapshot-handler.test.ts b/src/daemon/handlers/__tests__/snapshot-handler.test.ts index e71da15b5..0688307c9 100644 --- a/src/daemon/handlers/__tests__/snapshot-handler.test.ts +++ b/src/daemon/handlers/__tests__/snapshot-handler.test.ts @@ -7,7 +7,7 @@ import { handleSnapshotCommands } from '../snapshot.ts'; import { withSessionlessRunnerCleanup } from '../snapshot-session.ts'; import { captureSnapshot } from '../snapshot-capture.ts'; import { SessionStore } from '../../session-store.ts'; -import type { SessionState } from '../../types.ts'; +import type { DaemonResponse, SessionState } from '../../types.ts'; import { AppError } from '../../../kernel/errors.ts'; import { buildSnapshotSignatures } from '../../android-snapshot-freshness.ts'; import { buildInteractionSurfaceSignature } from '../../interaction-outcome-policy.ts'; @@ -484,6 +484,20 @@ function makeVersionedRefsScenario(sessionName: string) { return sessionStore; } +function expectInternalObservationResult(params: { + response: DaemonResponse | null | undefined; + session: SessionState | undefined; + publishedGeneration: number | undefined; + publishedTree: SessionState['snapshot']; +}): void { + expect(params.response?.ok).toBe(true); + expect(params.response?.ok ? params.response.data?.refsGeneration : undefined).toBeUndefined(); + expect(params.session?.snapshotGeneration).toBe((params.publishedGeneration as number) + 1); + expect(params.session?.snapshot).not.toBe(params.publishedTree); + expect(params.session?.refFrameGeneration).toBe(params.publishedGeneration); + expect(params.session?.refFrameTree).toBe(params.publishedTree); +} + test('snapshot responses carry refsGeneration and advance it per capture (#1076 versioned refs)', async () => { const sessionName = 'android-refs-generation'; const sessionStore = makeVersionedRefsScenario(sessionName); @@ -517,6 +531,43 @@ test('diff advances the generation without issuing refsGeneration (#1076 version expect(sessionStore.get(sessionName)?.snapshotGeneration).toBe(seed + 1); }); +test('daemon-private snapshot observation advances capture state without publishing ref authority', async () => { + const sessionName = 'android-internal-observation'; + const sessionStore = makeVersionedRefsScenario(sessionName); + + await runVersionedRefsCommand({ sessionStore, sessionName, command: 'snapshot' }); + const published = sessionStore.get(sessionName); + const publishedGeneration = published?.refFrameGeneration; + const publishedTree = published?.refFrameTree; + + mockDispatch.mockResolvedValue({ + nodes: [{ index: 0, depth: 0, type: 'android.widget.Button', label: 'Internal' }], + truncated: false, + backend: 'android', + }); + + const response = await handleSnapshotCommands({ + req: { + token: 't', + session: sessionName, + command: 'snapshot', + positionals: [], + flags: {}, + internal: { observationOnly: true }, + }, + sessionName, + logPath: '/tmp/daemon.log', + sessionStore, + }); + + expectInternalObservationResult({ + response, + session: sessionStore.get(sessionName), + publishedGeneration, + publishedTree, + }); +}); + test('snapshot surfaces filtered-to-zero Android guidance for interactive snapshots', async () => { const sessionStore = makeSessionStore(); const sessionName = 'android-empty-interactive'; @@ -1935,6 +1986,58 @@ test('wait text on iOS without app bundle id uses snapshot path', async () => { ); }); +test('wait text falls back to the canonical snapshot after an Apple runner miss', async () => { + const sessionStore = makeSessionStore(); + const sessionName = 'ios-wait-runner-miss'; + sessionStore.set(sessionName, { + ...makeSession(sessionName, iosSimulatorDevice), + appBundleId: 'com.example.app', + }); + + mockRunnerCommand.mockResolvedValue({ found: false }); + mockDispatch.mockResolvedValue({ + nodes: [ + { + index: 0, + depth: 0, + type: 'Window', + rect: { x: 0, y: 0, width: 390, height: 844 }, + }, + { + index: 1, + depth: 1, + parentIndex: 0, + type: 'StaticText', + label: 'Agent Device Tester', + rect: { x: 20, y: 80, width: 240, height: 40 }, + }, + ], + }); + + const response = await handleSnapshotCommands({ + req: { + token: 't', + session: sessionName, + command: 'wait', + positionals: ['Agent Device Tester', '5000'], + flags: {}, + }, + sessionName, + logPath: '/tmp/daemon.log', + sessionStore, + }); + + expect(response?.ok).toBe(true); + expect(mockRunnerCommand).toHaveBeenCalledTimes(1); + expect(mockDispatch).toHaveBeenCalledWith( + expect.anything(), + 'snapshot', + [], + undefined, + expect.anything(), + ); +}); + // fallow-ignore-next-line complexity test('wait selector uses direct iOS selector query when possible', async () => { const sessionStore = makeSessionStore(); diff --git a/src/daemon/handlers/session-open.ts b/src/daemon/handlers/session-open.ts index bded3efab..572f91645 100644 --- a/src/daemon/handlers/session-open.ts +++ b/src/daemon/handlers/session-open.ts @@ -592,9 +592,10 @@ async function openNewSessionWithAdvisoryClaim(params: { await rollbackNewSessionClaim(localClaim.ownership, effects); return details.response; } - // Preparation above is validation-only. `completeOpenCommand` can prewarm a runner, - // relaunch-close an app, or write runtime hints before its main open dispatch, so a - // failure from that point cannot prove the device is unchanged. + // Preparation can boot the device or warm caches, but it cannot establish session + // ownership. `completeOpenCommand` can relaunch-close an app or write runtime hints + // before its main open dispatch, so a failure from that point cannot prove ownership + // was not established. effects.mayHaveStarted = true; const response = await completeOpenCommand({ req, @@ -680,6 +681,7 @@ export async function handleOpenCommand(params: { } const device = await refreshSessionDeviceIfNeeded(session.device); + await req.internal?.retainDeviceExecutionLock?.(device.id); const details = await prepareOpenCommandDetails({ req, sessionName, @@ -740,6 +742,7 @@ export async function handleOpenCommand(params: { req.flags ?? {}, buildOpenTargetDeviceResolutionOptions(openTarget), ); + await req.internal?.retainDeviceExecutionLock?.(device.id); const surfaceResult = resolveOpenSurfaceResponse(device, req.flags?.surface, openTarget); if (typeof surfaceResult !== 'string') { return surfaceResult; diff --git a/src/daemon/handlers/session-replay-divergence-publication.ts b/src/daemon/handlers/session-replay-divergence-publication.ts new file mode 100644 index 000000000..3240df3a1 --- /dev/null +++ b/src/daemon/handlers/session-replay-divergence-publication.ts @@ -0,0 +1,126 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import type { ResponseLevel } from '../../kernel/contracts.ts'; +import { redactDiagnosticData } from '../../kernel/redaction.ts'; +import { boundReplayDivergence, type ReplayDivergence } from '../../replay/divergence.ts'; +import { + bindInternalObservationAuthority, + type InternalObservationEvidence, +} from '../internal-observation.ts'; +import { SessionStore } from '../session-store.ts'; + +/** + * Daemon-owned replay projection and publication boundary. The response or + * overflow artifact is projected first; ref authority is then activated + * synchronously from that exact successful projection. + */ +export function boundReplayDivergenceForSession(params: { + sessionStore: SessionStore; + sessionName: string; + divergence: ReplayDivergence; + responseLevel: ResponseLevel | undefined; + evidence: InternalObservationEvidence | undefined; + signal?: AbortSignal; +}): ReplayDivergence { + const { sessionStore, sessionName, divergence, responseLevel } = params; + let overflowProjection: ReplayDivergence | undefined; + let overflowArtifactPath: string | undefined; + const bounded = boundReplayDivergence({ + divergence, + level: responseLevel, + writeOverflowArtifact: (payload) => { + const artifactProjection = redactDiagnosticData(payload); + const result = writeReplayDivergenceArtifact(sessionStore, sessionName, artifactProjection); + if ('artifactPath' in result) { + overflowProjection = artifactProjection; + overflowArtifactPath = result.artifactPath; + } + return result; + }, + }); + const projection = overflowProjection ?? bounded; + const screen = projection.screen; + if (screen.state !== 'available' || screen.refs.length === 0) { + if (params.evidence) { + bindInternalObservationAuthority({ + sessionStore, + sessionName, + ...(params.signal ? { signal: params.signal } : {}), + }).finalize(params.evidence, { + refsGeneration: screen.state === 'available' ? screen.refsGeneration : undefined, + refs: [], + }); + } + return bounded; + } + if (!params.evidence) { + removeUnpublishedOverflowArtifact(overflowArtifactPath); + return suppressUnpublishedDivergenceRefs(bounded, 'missing-evidence'); + } + + const observationAuthority = bindInternalObservationAuthority({ + sessionStore, + sessionName, + ...(params.signal ? { signal: params.signal } : {}), + }); + const publication = observationAuthority.finalize(params.evidence, { + refsGeneration: screen.refsGeneration, + refs: screen.refs.map((entry) => entry.ref), + }); + if (publication.published === true) return bounded; + + removeUnpublishedOverflowArtifact(overflowArtifactPath); + return suppressUnpublishedDivergenceRefs(bounded, publication.reason); +} + +function suppressUnpublishedDivergenceRefs( + divergence: ReplayDivergence, + reason: 'missing-evidence' | 'empty' | 'cancelled' | 'stale-capture' | 'invalid-projection', +): ReplayDivergence { + const { overflow: _overflow, ...withoutOverflow } = divergence; + return { + ...withoutOverflow, + screen: { + state: 'unavailable', + reason: `ref-publication-${reason}`, + hint: 'The replay observation changed before its refs could be published. Take a new snapshot before targeting an element.', + }, + suggestions: divergence.suggestions.map(({ ref: _ref, ...suggestion }) => suggestion), + ...(divergence.targetBinding + ? { + targetBinding: { + ...divergence.targetBinding, + candidates: divergence.targetBinding.candidates.map( + ({ ref: _ref, ...candidate }) => candidate, + ), + }, + } + : {}), + }; +} + +function removeUnpublishedOverflowArtifact(artifactPath: string | undefined): void { + if (!artifactPath) return; + try { + fs.unlinkSync(artifactPath); + } catch { + // The unreturned path is not exposed to the caller. Cleanup is best effort. + } +} + +function writeReplayDivergenceArtifact( + sessionStore: SessionStore, + sessionName: string, + payload: ReplayDivergence, +): { artifactPath: string } | { artifactUnavailable: true } { + try { + const dir = path.join(sessionStore.ensureSessionDir(sessionName), 'replay-divergence'); + fs.mkdirSync(dir, { recursive: true }); + const fileName = `${Date.now()}-step${payload.step.index}.json`; + const artifactPath = path.join(dir, fileName); + fs.writeFileSync(artifactPath, `${JSON.stringify(payload, null, 2)}\n`); + return { artifactPath }; + } catch { + return { artifactUnavailable: true }; + } +} diff --git a/src/daemon/handlers/session-replay-divergence.ts b/src/daemon/handlers/session-replay-divergence.ts index 3ebf5ee18..b7a3c22ca 100644 --- a/src/daemon/handlers/session-replay-divergence.ts +++ b/src/daemon/handlers/session-replay-divergence.ts @@ -1,13 +1,9 @@ -import fs from 'node:fs'; -import path from 'node:path'; import { sleep } from '../../utils/timeouts.ts'; -import { markSessionPartialRefsIssued, setSessionSnapshot } from '../session-snapshot.ts'; import { isSparseSnapshotQualityVerdict, isUnreadableCaptureContentError, } from '../../snapshot/snapshot-quality.ts'; import { displayLabel, formatRole } from '../../snapshot/snapshot-lines.ts'; -import { redactDiagnosticData } from '../../kernel/redaction.ts'; import type { CommandFlags } from '../../core/dispatch.ts'; import type { DaemonError, ResponseLevel } from '../../kernel/contracts.ts'; import type { SnapshotNode } from '../../kernel/snapshot.ts'; @@ -28,11 +24,15 @@ import { type ReplayRepairHintCapture, } from './session-replay-repair-hint.ts'; import { SessionStore } from '../session-store.ts'; +import { + bindInternalObservationAuthority, + type InternalObservationEvidence, +} from '../internal-observation.ts'; +import { boundReplayDivergenceForSession } from './session-replay-divergence-publication.ts'; import type { ReplayReportAction } from './session-replay-report-action.ts'; import type { SessionAction, SessionState } from '../types.ts'; import { REPLAY_DIVERGENCE_SUGGESTION_LIMIT, - boundReplayDivergence, createReplayDivergenceSanitizer, type ReplayDivergence, type ReplayDivergenceScreen, @@ -68,6 +68,7 @@ export async function buildReplayFailureDivergence(params: { planActions: SessionAction[]; /** SHA-256 digest of the canonical plan `planActions` came from (`computeReplayPlanDigest`). */ planDigest: string; + signal?: AbortSignal; }): Promise { const { error, @@ -83,6 +84,7 @@ export async function buildReplayFailureDivergence(params: { scrubVars = [], planActions, planDigest, + signal, } = params; const sanitize = createReplayDivergenceSanitizer(scrubVars); @@ -146,27 +148,13 @@ export async function buildReplayFailureDivergence(params: { repairHint, }; - return boundReplayDivergenceForSession({ sessionStore, sessionName, divergence, responseLevel }); -} - -/** - * Shared response-level bounding + overflow-artifact wiring (`boundReplayDivergence` - * bound to this session's artifact directory). Exported so step 4's - * target-binding divergence goes through the exact same bounding/overflow - * behavior as an action-failure divergence. - */ -export function boundReplayDivergenceForSession(params: { - sessionStore: SessionStore; - sessionName: string; - divergence: ReplayDivergence; - responseLevel: ResponseLevel | undefined; -}): ReplayDivergence { - const { sessionStore, sessionName, divergence, responseLevel } = params; - return boundReplayDivergence({ + return boundReplayDivergenceForSession({ + sessionStore, + sessionName, divergence, - level: responseLevel, - writeOverflowArtifact: (payload) => - writeReplayDivergenceArtifact(sessionStore, sessionName, payload), + responseLevel, + evidence: observation.state === 'available' ? observation.evidence : undefined, + ...(signal ? { signal } : {}), }); } @@ -175,6 +163,7 @@ export type DivergenceObservation = state: 'available'; nodes: SnapshotNode[]; refsGeneration: number; + evidence: InternalObservationEvidence; /** Session's app bundle id at capture time; threaded to `buildDivergenceScreen`'s chrome filter (Android IME-scope guard — inert on iOS). */ appBundleId: string | undefined; } @@ -190,10 +179,10 @@ export function toReplayRepairHintCapture( } /** - * The single post-failure capture, blessed via the ADR-0014 partial ref-issuing - * sequence (setSessionSnapshot -> markSessionPartialRefsIssued -> store): a - * divergence screen publishes only its bounded ref set, so it activates a - * PARTIAL frame authorizing exactly those bodies, not a complete namespace. + * The single post-failure internal capture. It updates operational observation + * state and returns opaque lineage evidence without touching client ref + * authority. The daemon-owned response finalizer above activates a PARTIAL + * frame only after response-level bounding and overflow projection are exact. * Sparse captures do not write back (selector-capture reliability contract), * so a sparse verdict degrades the whole observation. * @@ -330,25 +319,17 @@ async function captureDivergenceObservationAttempt(params: { retryable: true, }; } - setSessionSnapshot(session, snapshot); - // ADR 0014 (#1257) + #1264: the divergence screen publishes exactly the - // ranked, occlusion-resolved, capped ref set `screen.refs` renders. Activate - // a PARTIAL frame authorizing precisely THOSE bodies — derived from the same - // `selectDivergenceScreenRefNodes` the digest uses, so the frame never - // authorizes a ref the screen hides (over-pin risk) nor rejects one the - // screen advertised (e.g. the mass-covered fallback surfaces covered refs - // that the old non-covered-only filter would have excluded here). - const digestBodies = selectDivergenceScreenRefNodes( - snapshot.nodes, - session.appBundleId, - ).nodes.map((node) => node.ref as string); - markSessionPartialRefsIssued(session, digestBodies); - sessionStore.set(sessionName, session); + const observationAuthority = bindInternalObservationAuthority({ + sessionStore, + sessionName, + }); + const stored = observationAuthority.store(snapshot); return { observation: { state: 'available', nodes: snapshot.nodes, - refsGeneration: session.snapshotGeneration ?? 0, + refsGeneration: stored.refsGeneration, + evidence: stored.evidence, appBundleId: session.appBundleId, }, retryable: false, @@ -460,8 +441,7 @@ function isForeignOverlayDismissTarget( * The single source of truth for which nodes a divergence `screen.refs` * publishes, and in what order. Both the rendered `screen.refs` digest * (`buildReplayDivergenceScreenRefs`) AND the ADR-0014 partial ref frame the - * capture authorizes (`captureDivergenceObservation` → - * `markSessionPartialRefsIssued`) derive from THIS function, so the authorized + * finalizer may authorize derive from THIS function, so the authorized * ref set is exactly the set the agent is shown — never a superset it can pin * refs outside of, nor a subset that rejects a ref the screen advertised. * Returns the capped node list plus whether ranking overflowed the cap. @@ -673,20 +653,3 @@ export function buildReplayDivergenceSuggestionForNode(params: { ...(label ? { label: sanitize(label) } : {}), }; } - -function writeReplayDivergenceArtifact( - sessionStore: SessionStore, - sessionName: string, - payload: ReplayDivergence, -): { artifactPath: string } | { artifactUnavailable: true } { - try { - const dir = path.join(sessionStore.ensureSessionDir(sessionName), 'replay-divergence'); - fs.mkdirSync(dir, { recursive: true }); - const fileName = `${Date.now()}-step${payload.step.index}.json`; - const artifactPath = path.join(dir, fileName); - fs.writeFileSync(artifactPath, `${JSON.stringify(redactDiagnosticData(payload), null, 2)}\n`); - return { artifactPath }; - } catch { - return { artifactUnavailable: true }; - } -} diff --git a/src/daemon/handlers/session-replay-maestro-failure.ts b/src/daemon/handlers/session-replay-maestro-failure.ts index a86903fd6..0f29d2e47 100644 --- a/src/daemon/handlers/session-replay-maestro-failure.ts +++ b/src/daemon/handlers/session-replay-maestro-failure.ts @@ -23,19 +23,20 @@ import { SessionStore } from '../session-store.ts'; import type { DaemonRequest, DaemonResponse, SessionState } from '../types.ts'; import type { ReplayReportAction } from './session-replay-report-action.ts'; import { - boundReplayDivergenceForSession, buildReplayDivergenceSuggestionForNode, buildDivergenceScreen, captureDivergenceObservation, toReplayRepairHintCapture, type DivergenceFieldSanitizer, } from './session-replay-divergence.ts'; +import { boundReplayDivergenceForSession } from './session-replay-divergence-publication.ts'; import { computeReplayRepairHint } from './session-replay-repair-hint.ts'; import { rankAndDedupeReplaySuggestions } from './session-replay-suggestion-ranking.ts'; import { buildReplayDivergenceFailureResponseFromDescriptor, hoistReplayFailureCauseDiagnosticMeta, } from './session-replay-runtime-failure-response.ts'; +import { getRequestSignal } from '../../request/cancel.ts'; export type MaestroFailedEngineEvent = MaestroEngineEvent & { readonly durationMs: number; @@ -85,6 +86,7 @@ export async function buildTypedMaestroFailureResponse(params: { readonly snapshotDiagnostics?: SnapshotDiagnosticsSummary; }): Promise { const { event, plan, replayPath, req, sessionName, sessionStore, logPath } = params; + const requestSignal = getRequestSignal(req.meta?.requestId); const report = buildTypedMaestroFailureReportProjection(event, req); const cause = hoistReplayFailureCauseDiagnosticMeta(params.error); const scrubVars = collectMaestroTextScrubVars(report.command); @@ -166,6 +168,8 @@ export async function buildTypedMaestroFailureResponse(params: { sessionName, divergence, responseLevel: req.meta?.responseLevel, + evidence: observation.state === 'available' ? observation.evidence : undefined, + ...(requestSignal ? { signal: requestSignal } : {}), }); return buildReplayDivergenceFailureResponseFromDescriptor({ error: safeCause, diff --git a/src/daemon/handlers/session-replay-runtime-failure.ts b/src/daemon/handlers/session-replay-runtime-failure.ts index b54fd085f..48e670bd0 100644 --- a/src/daemon/handlers/session-replay-runtime-failure.ts +++ b/src/daemon/handlers/session-replay-runtime-failure.ts @@ -11,6 +11,7 @@ import { buildReplayDivergenceFailureResponse, hoistReplayFailureCauseDiagnosticMeta, } from './session-replay-runtime-failure-response.ts'; +import { getRequestSignal } from '../../request/cancel.ts'; export async function withReplayFailureDiagnostics(params: { response: DaemonResponse; @@ -87,6 +88,7 @@ async function withReplayFailureContext(params: { scrubVars, planActions, planDigest, + signal: getRequestSignal(req.meta?.requestId), }); return buildReplayDivergenceFailureResponse({ error: cause, diff --git a/src/daemon/handlers/session-replay-runtime.ts b/src/daemon/handlers/session-replay-runtime.ts index cde7a6181..0712379ee 100644 --- a/src/daemon/handlers/session-replay-runtime.ts +++ b/src/daemon/handlers/session-replay-runtime.ts @@ -60,6 +60,7 @@ import { isTypedMaestroReplay, runTypedMaestroReplayFile, } from './session-replay-maestro-runtime.ts'; +import { getRequestSignal } from '../../request/cancel.ts'; /** Per-run invariants for a single replay step (ADR 0012 step 4 verify + dispatch + guard). */ type ReplayStepContext = { @@ -76,6 +77,7 @@ type ReplayStepContext = { actionTracePath: string | undefined; responseLevel: ResponseLevel | undefined; invoke: DaemonInvokeFn; + signal: AbortSignal | undefined; }; /** @@ -109,6 +111,7 @@ async function resolveReplayStepResponse( responseLevel: ctx.responseLevel, planActions: ctx.actions, planDigest: ctx.planDigest, + signal: ctx.signal, }); if (!verification.verified) return verification.response; const guard = verification.guard; @@ -178,6 +181,7 @@ async function convertIdentityRefusalResponse(params: { responseLevel: ctx.responseLevel, planActions: ctx.actions, planDigest: ctx.planDigest, + signal: ctx.signal, }; if (params.guard && isReplayTargetGuardMismatchResponse(response)) { return await buildReplayTargetGuardMismatchResponse({ ...mismatchParams, guard: params.guard }); @@ -252,6 +256,7 @@ export async function runReplayScriptFile(params: { actionTracePath, responseLevel: req.meta?.responseLevel, invoke, + signal: getRequestSignal(req.meta?.requestId), }; const failure = await executeReplayActions({ req, diff --git a/src/daemon/handlers/session-replay-target-verification.ts b/src/daemon/handlers/session-replay-target-verification.ts index ca2da1d7f..388144203 100644 --- a/src/daemon/handlers/session-replay-target-verification.ts +++ b/src/daemon/handlers/session-replay-target-verification.ts @@ -29,15 +29,16 @@ import { resolveTargetIdentityVerification } from '../../core/command-descriptor import { parseWaitPositionals } from '../../core/wait-positionals.ts'; import type { DaemonResponse, SessionAction } from '../types.ts'; import { SessionStore } from '../session-store.ts'; +import type { InternalObservationEvidence } from '../internal-observation.ts'; import { boundedLocalIdentity } from '../session-target-evidence.ts'; import { tryParseSelectorChain } from '../../selectors/index.ts'; import { buildDivergenceScreen, - boundReplayDivergenceForSession, captureDivergenceObservation, resolveSuggestionMatchingConfig, toReplayRepairHintCapture, } from './session-replay-divergence.ts'; +import { boundReplayDivergenceForSession } from './session-replay-divergence-publication.ts'; import { computeReplayRepairHint, type ReplayRepairHintCapture, @@ -100,6 +101,7 @@ type TargetBindingDivergenceContext = { /** ADR 0012 step 5: the full top-level plan + its digest, for `resume`. */ planActions: SessionAction[]; planDigest: string; + signal?: AbortSignal; }; type TargetBindingDivergenceBuilt = { @@ -112,6 +114,7 @@ type TargetBindingDivergenceBuilt = { causeMessage: string; causeHint?: string; screen: ReplayDivergence['screen']; + publicationEvidence?: InternalObservationEvidence; /** ADR 0012 decision 6, R3: the same capture `screen` was built from, for the `repairHint` container test. */ repairCapture: ReplayRepairHintCapture; }; @@ -191,6 +194,8 @@ function buildTargetBindingDivergenceResponse( sessionName, divergence, responseLevel, + evidence: built.publicationEvidence, + ...(context.signal ? { signal: context.signal } : {}), }); const cause: DaemonError = { code: built.causeCode, message: built.causeMessage }; return buildReplayDivergenceFailureResponse({ @@ -204,7 +209,7 @@ function buildTargetBindingDivergenceResponse( }); } -export async function verifyReplayActionTarget(params: { +type ReplayTargetDivergenceParams = { action: SessionAction; scope: ReplayVarScope; sourcePath: string; @@ -218,7 +223,12 @@ export async function verifyReplayActionTarget(params: { responseLevel: ResponseLevel | undefined; planActions: SessionAction[]; planDigest: string; -}): Promise { + signal?: AbortSignal; +}; + +export async function verifyReplayActionTarget( + params: ReplayTargetDivergenceParams, +): Promise { const { action, scope, @@ -233,6 +243,7 @@ export async function verifyReplayActionTarget(params: { responseLevel, planActions, planDigest, + signal, } = params; const recorded = action.targetEvidence; @@ -263,6 +274,7 @@ export async function verifyReplayActionTarget(params: { scrubVars, planActions, planDigest, + signal, }; const buildRecordedUnverifiableResponse = async (): Promise => { // Decision 3 path 1: a recorded-`unverifiable` annotation fires before @@ -284,6 +296,7 @@ export async function verifyReplayActionTarget(params: { causeMessage: 'The recorded target evidence could not verify itself when it was captured (a structural capture anomaly), so replay cannot trust it before acting.', screen: buildDivergenceScreen(observation, sanitize), + publicationEvidence: publicationEvidenceFrom(observation), repairCapture: toReplayRepairHintCapture(observation), }); }; @@ -388,6 +401,7 @@ export async function verifyReplayActionTarget(params: { causeCode: classification.causeCode, causeMessage: classification.causeMessage, screen: buildDivergenceScreen(observation, sanitize), + publicationEvidence: observation.evidence, repairCapture: toReplayRepairHintCapture(observation), }), }; @@ -408,21 +422,8 @@ export function isReplayTargetGuardMismatchResponse(response: DaemonResponse): b return !response.ok && response.error.details?.reason === REPLAY_TARGET_GUARD_MISMATCH_REASON; } -type PostDispatchMismatchParams = { - action: SessionAction; - scope: ReplayVarScope; +type PostDispatchMismatchParams = ReplayTargetDivergenceParams & { failedResponse: DaemonResponse; - sourcePath: string; - sourceLine: number; - replayPath: string; - step: number; - sessionName: string; - sessionStore: SessionStore; - logPath: string; - artifactPaths: string[]; - responseLevel: ResponseLevel | undefined; - planActions: SessionAction[]; - planDigest: string; }; type PostDispatchMismatchEvidence = { @@ -480,6 +481,7 @@ async function buildPostDispatchIdentityMismatchResponse( scrubVars, planActions: params.planActions, planDigest: params.planDigest, + signal: params.signal, }, { kind: 'identity-mismatch', @@ -490,11 +492,18 @@ async function buildPostDispatchIdentityMismatchResponse( causeCode: 'IDENTITY_MISMATCH', causeMessage: evidence.causeMessage, screen: buildDivergenceScreen(observation, sanitize), + publicationEvidence: publicationEvidenceFrom(observation), repairCapture: toReplayRepairHintCapture(observation), }, ); } +function publicationEvidenceFrom( + observation: Awaited>, +): InternalObservationEvidence | undefined { + return observation.state === 'available' ? observation.evidence : undefined; +} + export async function buildReplayTargetGuardMismatchResponse( params: PostDispatchMismatchParams & { guard: ReplayVerifiedTargetGuard }, ): Promise { diff --git a/src/daemon/internal-observation.ts b/src/daemon/internal-observation.ts new file mode 100644 index 000000000..c5bca9195 --- /dev/null +++ b/src/daemon/internal-observation.ts @@ -0,0 +1,211 @@ +import type { SnapshotState } from '../kernel/snapshot.ts'; +import { readSessionRuntimeRevision } from './ref-frame.ts'; +import { markSessionPartialRefsIssued, setSessionSnapshot } from './session-snapshot.ts'; +import { SessionStore } from './session-store.ts'; +import type { SessionState } from './types.ts'; + +declare const INTERNAL_OBSERVATION_EVIDENCE: unique symbol; + +/** + * Opaque capture lineage that engines may carry as data but cannot use to + * publish refs. Only the daemon-owned finalizer in this module can resolve it. + */ +export type InternalObservationEvidence = { + readonly [INTERNAL_OBSERVATION_EVIDENCE]: true; +}; + +type RefFrameLineage = Readonly<{ + state: SessionState['refFrameState']; + scope: SessionState['refFrameScope']; + tree: SessionState['refFrameTree']; + generation: SessionState['refFrameGeneration']; +}>; + +type InternalObservationLineage = Readonly<{ + sessionName: string; + session: SessionState; + sessionCreatedAt: number; + snapshot: SnapshotState; + snapshotGeneration: number; + runtimeRevision: number; + refFrame: RefFrameLineage; +}>; + +const evidenceLineage = new WeakMap(); + +type StoredInternalObservation = Readonly<{ + evidence: InternalObservationEvidence; + refsGeneration: number; +}>; + +type ClientRefPublicationProjection = Readonly<{ + refsGeneration: number | undefined; + refs: readonly string[]; +}>; + +type ClientRefPublicationResult = + | Readonly<{ published: true; refsGeneration: number; refCount: number }> + | Readonly<{ + published: false; + reason: 'empty' | 'cancelled' | 'stale-capture' | 'invalid-projection'; + }>; + +type InternalObservationAuthority = Readonly<{ + store(snapshot: SnapshotState): StoredInternalObservation; + finalize( + evidence: InternalObservationEvidence, + projection: ClientRefPublicationProjection, + ): ClientRefPublicationResult; +}>; + +type BoundInternalObservationSession = Readonly<{ + sessionStore: SessionStore; + sessionName: string; + signal?: AbortSignal; +}>; + +/** + * Bind observation authority to one already admitted, locked session. Callers + * cannot address another session through the resulting capability, and engines + * receive only opaque evidence values rather than this authority. + */ +export function bindInternalObservationAuthority( + params: BoundInternalObservationSession, +): InternalObservationAuthority { + return { + store: (snapshot) => storeInternalObservation(params, snapshot), + finalize: (evidence, projection) => + finalizeClientRefPublication({ + ...params, + evidence, + projection, + }), + }; +} + +/** + * Daemon-private capture finalization: update operational observation state + * without activating, replacing, or expiring client ref authority. + */ +function storeInternalObservation( + params: Pick, + snapshot: SnapshotState, +): StoredInternalObservation { + const { sessionStore, sessionName } = params; + const session = sessionStore.get(sessionName); + if (!session) { + throw new Error('Internal observation session is no longer available.'); + } + setSessionSnapshot(session, snapshot); + sessionStore.set(sessionName, session); + const snapshotGeneration = session.snapshotGeneration; + if (snapshotGeneration === undefined) { + throw new Error('Internal observation did not establish a snapshot generation.'); + } + + const evidence = {} as InternalObservationEvidence; + evidenceLineage.set(evidence, { + sessionName, + session, + sessionCreatedAt: session.createdAt, + snapshot, + snapshotGeneration, + runtimeRevision: readSessionRuntimeRevision(session), + refFrame: readRefFrameLineage(session), + }); + return { evidence, refsGeneration: snapshotGeneration }; +} + +/** + * Paired synchronous publication finalizer. It activates exactly the refs + * exposed by the already-projected inline response or successfully written + * overflow artifact, and only while capture lineage is still current. + * + * No asynchronous work may occur after this returns `published: true` and + * before the response returns to the client. + */ +function finalizeClientRefPublication(params: { + sessionStore: SessionStore; + sessionName: string; + evidence: InternalObservationEvidence; + projection: ClientRefPublicationProjection; + signal?: AbortSignal; +}): ClientRefPublicationResult { + const lineage = evidenceLineage.get(params.evidence); + // Evidence is a one-shot capability. Consume it before every outcome, + // including empty, cancelled, invalid, and stale attempts, so request-end + // finalization can never be retried into publication. + evidenceLineage.delete(params.evidence); + const refs = normalizeRefBodies(params.projection.refs); + if (refs.size === 0) return { published: false, reason: 'empty' }; + if (params.signal?.aborted === true) return { published: false, reason: 'cancelled' }; + + if (!lineage || !isCurrentLineage(params, lineage)) { + return { published: false, reason: 'stale-capture' }; + } + if ( + params.projection.refsGeneration !== lineage.snapshotGeneration || + !areProjectedRefsFromSnapshot(refs, lineage.snapshot) + ) { + return { published: false, reason: 'invalid-projection' }; + } + + markSessionPartialRefsIssued(lineage.session, refs); + return { + published: true, + refsGeneration: lineage.snapshotGeneration, + refCount: refs.size, + }; +} + +function isCurrentLineage( + params: Pick, + lineage: InternalObservationLineage, +): boolean { + const current = params.sessionStore.get(params.sessionName); + return ( + params.sessionName === lineage.sessionName && + current === lineage.session && + current.createdAt === lineage.sessionCreatedAt && + current.snapshot === lineage.snapshot && + current.snapshotGeneration === lineage.snapshotGeneration && + readSessionRuntimeRevision(current) === lineage.runtimeRevision && + sameRefFrameLineage(readRefFrameLineage(current), lineage.refFrame) + ); +} + +function readRefFrameLineage(session: SessionState): RefFrameLineage { + return { + state: session.refFrameState, + scope: session.refFrameScope, + tree: session.refFrameTree, + generation: session.refFrameGeneration, + }; +} + +function sameRefFrameLineage(left: RefFrameLineage, right: RefFrameLineage): boolean { + return ( + left.state === right.state && + left.scope === right.scope && + left.tree === right.tree && + left.generation === right.generation + ); +} + +function normalizeRefBodies(refs: readonly string[]): Set { + const normalized = new Set(); + for (const ref of refs) { + const withoutAt = ref.startsWith('@') ? ref.slice(1) : ref; + const suffix = withoutAt.indexOf('~'); + const body = suffix === -1 ? withoutAt : withoutAt.slice(0, suffix); + if (body.length > 0) normalized.add(body); + } + return normalized; +} + +function areProjectedRefsFromSnapshot(refs: ReadonlySet, snapshot: SnapshotState): boolean { + const capturedRefs = new Set( + snapshot.nodes.flatMap((node) => (typeof node.ref === 'string' ? [node.ref] : [])), + ); + return [...refs].every((ref) => capturedRefs.has(ref)); +} diff --git a/src/daemon/ref-frame.ts b/src/daemon/ref-frame.ts index f311a91fc..b2b664b02 100644 --- a/src/daemon/ref-frame.ts +++ b/src/daemon/ref-frame.ts @@ -1,5 +1,7 @@ import type { SessionState } from './types.ts'; +const runtimeRevisions = new WeakMap(); + /** * ADR 0014 session ref-frame lifetime — the authorization model for mutation * refs, kept distinct from the latest operational observation (`session.snapshot`). @@ -67,10 +69,28 @@ export function refFrameEpoch(session: SessionState): number | undefined { * side effect. */ export function expireRefFrame(session: SessionState): void { + advanceSessionRuntimeRevision(session); session.refFrameState = 'expired'; session.snapshotScopeSource = undefined; } +/** + * Monotonic, daemon-private revision for side-effect lineage. Unlike the + * client-visible snapshot/ref generations, this advances for every possible + * device mutation, including another mutation while the ref frame is already + * expired. Internal observation evidence uses it to refuse publication after + * any intervening side-effect seam. + */ +function advanceSessionRuntimeRevision(session: SessionState): number { + const next = readSessionRuntimeRevision(session) + 1; + runtimeRevisions.set(session, next); + return next; +} + +export function readSessionRuntimeRevision(session: SessionState): number { + return runtimeRevisions.get(session) ?? 0; +} + /** * Re-authorize a complete frame with scope `all` (ADR 0014). This is the only * transition that restores plain-ref mutation after an expiry, and it is diff --git a/src/daemon/request-binding.ts b/src/daemon/request-binding.ts index f1db4bcc1..574aa2549 100644 --- a/src/daemon/request-binding.ts +++ b/src/daemon/request-binding.ts @@ -21,7 +21,10 @@ export async function resolveRequestExecutionLockKeys(params: { const { req, sessionName, sessionStore } = params; const existingSession = sessionStore.get(sessionName); if (existingSession) { - return [deviceExecutionLockKey(existingSession.device.id)]; + return orderRequestExecutionLockKeys([ + sessionExecutionLockKey(sessionName), + deviceExecutionLockKey(existingSession.device.id), + ]); } const keys = new Set([sessionExecutionLockKey(sessionName)]); diff --git a/src/daemon/request-execution-locks.ts b/src/daemon/request-execution-locks.ts new file mode 100644 index 000000000..99f8c1e46 --- /dev/null +++ b/src/daemon/request-execution-locks.ts @@ -0,0 +1,100 @@ +import { withKeyedLock } from '../utils/keyed-lock.ts'; +import type { RequestExecutionLockKey } from './request-binding.ts'; + +type RetainedExecutionLock = { + acquired: Promise; + completion: Promise; + release(): void; +}; + +export type RequestExecutionLocks = { + run(task: () => Promise): Promise; + retainDevice(deviceId: string): Promise; +}; + +/** + * Owns the stable execution locks for one request and any device lock discovered + * after admission. Dynamic device retention is used by a fresh replay whose + * first open cannot select its device during advisory preflight. + */ +export function createRequestExecutionLocks(params: { + locks: Map>; + initialKeys: RequestExecutionLockKey[]; +}): RequestExecutionLocks { + const { locks, initialKeys } = params; + const initialKeySet = new Set(initialKeys); + const retainedLocks = new Map(); + let running = false; + + const releaseRetainedLocks = async (): Promise => { + const held = [...retainedLocks.values()]; + for (const lock of held) lock.release(); + await Promise.all(held.map(async (lock) => await lock.completion)); + retainedLocks.clear(); + }; + + return { + run: async (task) => { + if (running) { + throw new Error('Request execution locks cannot run the same scope concurrently.'); + } + running = true; + try { + return await withRequestExecutionLocks(locks, initialKeys, task); + } finally { + try { + await releaseRetainedLocks(); + } finally { + running = false; + } + } + }, + retainDevice: async (deviceId) => { + if (!running) { + throw new Error('A device execution lock can only be retained by a running request.'); + } + const key = `device:${deviceId}` as const; + if (initialKeySet.has(key)) return; + + let retained = retainedLocks.get(key); + if (!retained) { + retained = retainExecutionLock(locks, key); + retainedLocks.set(key, retained); + } + await retained.acquired; + }, + }; +} + +async function withRequestExecutionLocks( + locks: Map>, + keys: RequestExecutionLockKey[], + task: () => Promise, +): Promise { + const [key, ...remainingKeys] = keys; + if (!key) return await task(); + return await withKeyedLock( + locks, + key, + async () => await withRequestExecutionLocks(locks, remainingKeys, task), + ); +} + +function retainExecutionLock( + locks: Map>, + key: RequestExecutionLockKey, +): RetainedExecutionLock { + let markAcquired: () => void = () => {}; + let release: () => void = () => {}; + const acquired = new Promise((resolve) => { + markAcquired = resolve; + }); + const released = new Promise((resolve) => { + release = resolve; + }); + const completion = withKeyedLock(locks, key, async () => { + markAcquired(); + await released; + }); + return { acquired, completion, release }; +} diff --git a/src/daemon/request-execution-scope.ts b/src/daemon/request-execution-scope.ts index 2c4fe9b17..5b8296fb7 100644 --- a/src/daemon/request-execution-scope.ts +++ b/src/daemon/request-execution-scope.ts @@ -1,6 +1,5 @@ import type { CommandFlags } from '../core/dispatch.ts'; import type { DaemonArtifactType } from '../kernel/contracts.ts'; -import { withKeyedLock } from '../utils/keyed-lock.ts'; import { emitDiagnostic, getDiagnosticsMeta, @@ -18,11 +17,8 @@ import { assertLockedLeaseAdmissionPreflight, cleanupExpiredLeasedSession, } from './lease-lifecycle.ts'; -import { - prepareLockedRequestBinding, - resolveRequestExecutionLockKeys, - type RequestExecutionLockKey, -} from './request-binding.ts'; +import { prepareLockedRequestBinding, resolveRequestExecutionLockKeys } from './request-binding.ts'; +import { createRequestExecutionLocks } from './request-execution-locks.ts'; import { throwIfRequestCanceled } from '../request/cancel.ts'; import { finalizeDaemonResponse } from './request-finalization.ts'; import { refreshRecordingHealth } from './request-recording-health.ts'; @@ -58,6 +54,7 @@ export type RequestExecutionScope = { startedAtMs: number; runAdmitted(task: () => Promise): Promise; runLocked(task: () => Promise): Promise; + retainDeviceExecutionLock(deviceId: string): Promise; throwIfCanceled(): void; }; @@ -66,6 +63,7 @@ export type LockedRequestScope = { sessionName: string; logPath: string; existingSession: SessionState | undefined; + retainDeviceExecutionLock(deviceId: string): Promise; finalize(response: DaemonResponse): DaemonResponse; contextFromFlags( flags: CommandFlags | undefined, @@ -133,6 +131,10 @@ export async function createRequestExecutionScope(params: { ? await resolveRequestExecutionLockKeys({ req: scopedReq, sessionName, sessionStore }) : []; const executionLocks = getLeaseRegistryExecutionLocks(leaseRegistry); + const requestExecutionLocks = createRequestExecutionLocks({ + locks: executionLocks, + initialKeys: executionLockKeys, + }); const scope: RequestExecutionScope = { req: scopedReq, @@ -141,6 +143,8 @@ export async function createRequestExecutionScope(params: { requestLogPath, runnerLogPath, startedAtMs, + retainDeviceExecutionLock: async (deviceId) => + await requestExecutionLocks.retainDevice(deviceId), throwIfCanceled: () => throwIfRequestCanceled(scopedReq.meta?.requestId), runAdmitted: async (task) => { throwIfRequestCanceled(scopedReq.meta?.requestId); @@ -161,12 +165,7 @@ export async function createRequestExecutionScope(params: { }, runLocked: async (task) => { throwIfRequestCanceled(scopedReq.meta?.requestId); - if (executionLockKeys.length === 0) return await scope.runAdmitted(task); - return await withRequestExecutionLocks( - executionLocks, - executionLockKeys, - async () => await scope.runAdmitted(task), - ); + return await requestExecutionLocks.run(async () => await scope.runAdmitted(task)); }, }; return scope; @@ -191,20 +190,6 @@ export async function createRequestExecutionScope(params: { } } -async function withRequestExecutionLocks( - locks: Map>, - keys: RequestExecutionLockKey[], - task: () => Promise, -): Promise { - const [key, ...remainingKeys] = keys; - if (!key) return await task(); - return await withKeyedLock( - locks, - key, - async () => await withRequestExecutionLocks(locks, remainingKeys, task), - ); -} - function applyRequestCommandDefaults(req: DaemonRequest): DaemonRequest { const flags = { ...(req.flags ?? {}) }; const changed = applyCommandDefaults(req.command, flags); @@ -295,6 +280,7 @@ export function prepareLockedRequestScope(params: { sessionName: scope.sessionName, logPath, existingSession, + retainDeviceExecutionLock: scope.retainDeviceExecutionLock, finalize, contextFromFlags, handlerContextFromFlags: (flags, appBundleId, traceLogPath) => diff --git a/src/daemon/request-router.ts b/src/daemon/request-router.ts index 497541c4b..22568f1ea 100644 --- a/src/daemon/request-router.ts +++ b/src/daemon/request-router.ts @@ -258,7 +258,17 @@ export function createRequestHandler(deps: RequestRouterDeps): DaemonInvokeFn { let childScope: RequestExecutionScope | undefined; try { - childScope = await createRequestExecutionScope({ req, sessionStore, leaseRegistry }); + const scopedReq = bindReplayDeviceExecutionLock(req, parentScope); + childScope = await createRequestExecutionScope({ + req: scopedReq, + sessionStore, + leaseRegistry, + }); + // The outer replay keeps its stable session lock plus the device lock + // from the first device binding through response projection and ref + // finalization. A same-session replay action reuses that admitted scope + // instead of reacquiring the non-reentrant locks. Nested changes remain + // visible to capture lineage through snapshot/frame/runtime/store state. return childScope.sessionName === parentScope.sessionName ? await executeRequestScope(childScope, providerScope) : await executeRequestScope(childScope); @@ -351,6 +361,24 @@ async function dispatchGenericForLockedScope(params: { return lockedScope.finalize(dispatchResponse); } +function bindReplayDeviceExecutionLock( + req: DaemonRequest, + parentScope: LockedRequestScope, +): DaemonRequest { + if (req.command !== 'open') return req; + const retainDeviceExecutionLock = req.internal?.retainDeviceExecutionLock; + return { + ...req, + internal: { + ...req.internal, + retainDeviceExecutionLock: async (deviceId) => { + await parentScope.retainDeviceExecutionLock(deviceId); + await retainDeviceExecutionLock?.(deviceId); + }, + }, + }; +} + function canRunReplayActionInCurrentScope( req: DaemonRequest, parentScope: LockedRequestScope, diff --git a/src/daemon/selector-runtime-backend.ts b/src/daemon/selector-runtime-backend.ts index 4b54313e9..6cb9c997c 100644 --- a/src/daemon/selector-runtime-backend.ts +++ b/src/daemon/selector-runtime-backend.ts @@ -174,7 +174,9 @@ async function findText(params: SelectorRuntimeDeviceParams, text: string): Prom const macosSurfaceResult = await findTextInMacosNonAppSurface(params, text); if (macosSurfaceResult !== null) return macosSurfaceResult; const appleRunnerResult = await findTextWithAppleRunner(params, text); - if (appleRunnerResult !== null) return appleRunnerResult; + // The runner query is a fast path, not the semantic source of truth. XCTest can report a + // transient miss for visible SwiftUI text that the canonical snapshot already contains. + if (appleRunnerResult === true) return true; return await findTextInWaitSnapshot(params, text); } diff --git a/src/daemon/server/transport.ts b/src/daemon/server/transport.ts index 86b5ed649..52d315e6e 100644 --- a/src/daemon/server/transport.ts +++ b/src/daemon/server/transport.ts @@ -63,7 +63,7 @@ export function createSocketServer(handleRequest: DaemonInvokeFn): DaemonServer let requestAbortRegistration: ReturnType; let streamProgress = false; try { - const req = JSON.parse(line) as DaemonRequest; + const req = parseSocketDaemonRequest(line); streamProgress = shouldStreamRequestProgress(req); requestIdForCleanup = resolveRequestTrackingId(req.meta?.requestId, 'socket'); req.meta = { @@ -113,6 +113,19 @@ export function createSocketServer(handleRequest: DaemonInvokeFn): DaemonServer return server; } +function parseSocketDaemonRequest(line: string): DaemonRequest { + const parsed = JSON.parse(line) as unknown; + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + return parsed as DaemonRequest; + } + // `internal` carries daemon-issued capabilities and provenance. The socket is + // a public transport like HTTP, so a token-bearing client must not be able to + // forge those semantics even though the legacy socket request otherwise + // preserves its existing raw wire shape. + const { internal: _internal, ...request } = parsed as Record; + return request as DaemonRequest; +} + export function listenNetServer(server: net.Server): Promise { return listenLoopbackServer(server, 'Failed to bind socket server'); } diff --git a/src/daemon/snapshot-runtime.ts b/src/daemon/snapshot-runtime.ts index a5d00a00b..9afa3c5ca 100644 --- a/src/daemon/snapshot-runtime.ts +++ b/src/daemon/snapshot-runtime.ts @@ -47,7 +47,7 @@ export async function dispatchSnapshotViaRuntime(params: { // the node tree itself stays plain `e12` refs (token economy). The // capture above already stored the next session via setRecord, so the // store holds the generation these refs were minted from. - const refsGeneration = params.sessionStore.get(sessionName)?.snapshotGeneration; + const refsGeneration = publishedSnapshotGeneration(req, params.sessionStore.get(sessionName)); // ADR 0014: retain provenance in the immutable operational/ref-frame tree; // project only the published copy so settle and replay keep the full evidence. const publicNodes = stripAndroidSystemChromeProvenance(result.nodes); @@ -65,6 +65,13 @@ export async function dispatchSnapshotViaRuntime(params: { }); } +function publishedSnapshotGeneration( + req: DaemonRequest, + session: SessionState | undefined, +): number | undefined { + return req.internal?.observationOnly === true ? undefined : session?.snapshotGeneration; +} + export async function dispatchSnapshotDiffViaRuntime(params: { req: DaemonRequest; sessionName: string; @@ -227,7 +234,8 @@ function createSnapshotRuntime(params: { // Only the snapshot command's response carries every stored node's // ref back to the client; diff returns a summary, so its refreshed // tree leaves client refs stale (#1076). - issuesRefsToClient: req.command === 'snapshot', + issuesRefsToClient: + req.command === 'snapshot' && req.internal?.observationOnly !== true, }), ); }, diff --git a/src/daemon/types.ts b/src/daemon/types.ts index a480ec4c8..776ca2bdf 100644 --- a/src/daemon/types.ts +++ b/src/daemon/types.ts @@ -50,7 +50,18 @@ export type DaemonOpenLifecycle = { type DaemonRequestInternal = { openLifecycle?: DaemonOpenLifecycle; + /** + * Request-owned capability used when a fresh replay discovers its device + * only inside the first open. The router retains that device's execution + * lock before dispatch and releases it after the outer replay finalizes. + */ + retainDeviceExecutionLock?: (deviceId: string) => Promise; admittedLease?: DeviceLease; + /** + * Daemon-composed hierarchy capture used as operational evidence only. + * It must not issue or replace client ref authority. + */ + observationOnly?: true; /** * Implicit caller scope resolved before a nested dispatch replaces the * public session name with its effective scoped key. diff --git a/test/integration/provider-scenarios/daemon-transport.test.ts b/test/integration/provider-scenarios/daemon-transport.test.ts index dd5b46892..76609be10 100644 --- a/test/integration/provider-scenarios/daemon-transport.test.ts +++ b/test/integration/provider-scenarios/daemon-transport.test.ts @@ -43,12 +43,21 @@ test('Provider-backed integration daemon socket transport frames requests and no command: 'session_list', positionals: [], meta: { requestId: 'req-socket-1' }, + internal: { + observationOnly: true, + replayPlanStep: true, + }, }), '{not-json}', ]); assert.equal(observedRequests.length, 1); assert.equal(observedRequests[0]?.meta?.requestId, 'req-socket-1'); + assert.equal( + observedRequests[0]?.internal, + undefined, + 'socket clients must not forge daemon-internal request capabilities', + ); assert.deepEqual(responses[0], { ok: true, data: {