From 8f5fb1cb5d01c403c4632be60eebbb60210ee40a Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 15:06:09 -0600 Subject: [PATCH] test(runner): verify Stop and authorization during held text writes --- .../src/native-session-runtime.test.ts | 236 ++++++++++++++++++ 1 file changed, 236 insertions(+) diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index c9a0b0510c5..8a57de5d07f 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -295,6 +295,242 @@ describe("executeNativeSession recovery", () => { expect(observed).toEqual(textEvents); }); + it("dispatches Stop through the published session while simulated persistence holds a full text collector", async () => { + const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; + let releaseFirst!: () => void; + const firstReleased = new Promise(resolve => { releaseFirst = resolve; }); + let collectorFull!: () => void; + const fullCollector = new Promise(resolve => { collectorFull = resolve; }); + const textEvents = Array.from({ length: 130 }, (_, index) => { + const event = { ...runnerEvent(index + 1, "item.delta", { kind: "agentMessage", text: "" }), priority: 1 as const }; + // Exactly 1 KiB per serialized envelope fills both collector bounds. + event.payload.text = "x".repeat(1024 - Buffer.byteLength(JSON.stringify(event))); + return event; + }); + const source = [...textEvents, runnerEvent(131, "run.result.proposed", result), runnerEvent(132, "turn.completed")]; + const stopError = new Error("simulated append authority revoked by Stop"); + let revoked = false; + let sourceReads = 0; + const cancel = vi.fn>(() => { + // Model the synchronous authority transition, independently of the + // deliberately held database receipt. Real PostgreSQL revocation is + // exercised separately by the control-plane port's held-row-lock tests. + revoked = true; + return { cleanup: Promise.resolve() }; + }); + const sessionResult = vi.fn(async () => ({ result, terminal, turnId: "turn-recovery" })); + const close = vi.fn(async () => undefined); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return capabilities; }, + async *events() { + for (const event of source) { + sourceReads += 1; + if (sourceReads === 65) collectorFull(); + yield event; + } + }, + async startTurn() { return { turnId: "turn-recovery" }; }, + cancel, + result: sessionResult, + async snapshot() { + return { backendKind: "mock", sessionId: "driver-recovery", identity, providerSessionId: "provider-recovery", + cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; + }, + close, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "blocked-stop-fixture", version: "1", capabilities }; }, + async openSession() { return session; }, + }; + const committed: PrpEvent[] = []; + const makeReceipt = (event: PrpEvent) => ({ cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, + disposition: "committed" as const }); + const appendEvents = vi.fn>(async events => { + if (revoked) throw stopError; + committed.push(...events); + return events.map(makeReceipt); + }); + const completeRun = vi.fn(async () => undefined); + const port: ControlPlanePort = { + async openRun() {}, async checkpointSession() {}, + async appendEvent(event, options) { + const prp = event as PrpEvent; + if (prp === textEvents[0]) await firstReleased; + options?.signal.throwIfAborted(); + if (revoked) throw stopError; + committed.push(prp); + return makeReceipt(prp); + }, + appendEvents, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + completeRun, + }; + const observed = vi.fn(() => null); + const published: { current: NativeSession | null } = { current: null }; + const execution = executeNativeSession({ input, backend, controlPlane: port, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + onSession(current) { published.current = current; }, resolveGovernedWait: observed }); + void execution.catch(() => undefined); + try { + await fullCollector; + await Promise.resolve(); + await Promise.resolve(); + expect(sourceReads).toBe(65); // One in flight and 64 buffered events. + expect(textEvents.slice(1, 65).reduce((bytes, event) => bytes + Buffer.byteLength(JSON.stringify(event)), 0)) + .toBe(64 * 1024); + expect(published.current).toBe(session); + expect(committed).toEqual([]); + expect(observed).not.toHaveBeenCalled(); + + // The real runtime publication supplies the control handle. Neither a + // full collector nor its pending append participates in Stop dispatch. + const cancellation = published.current!.cancel!({ reason: "User pressed Stop", signal: new AbortController().signal }); + expect(cancel).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ reason: "User pressed Stop" })); + await cancellation.cleanup; + expect(sourceReads).toBe(65); + expect(appendEvents).not.toHaveBeenCalled(); + expect(observed).not.toHaveBeenCalled(); + expect(sessionResult).not.toHaveBeenCalled(); + expect(completeRun).not.toHaveBeenCalled(); + + releaseFirst(); + await expect(execution).rejects.toThrow(stopError); + expect(committed).toEqual([]); + expect(sourceReads).toBe(65); + expect(appendEvents).not.toHaveBeenCalled(); + expect(observed).not.toHaveBeenCalled(); + expect(sessionResult).not.toHaveBeenCalled(); + expect(completeRun).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledOnce(); + } finally { + releaseFirst(); + await execution.catch(() => undefined); + } + }); + + it("keeps an explicit authorization behind queued text and its own simulated durable receipt", async () => { + const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; + let releaseFirst!: () => void; + const firstReleased = new Promise(resolve => { releaseFirst = resolve; }); + let releaseBatch!: () => void; + const batchReleased = new Promise(resolve => { releaseBatch = resolve; }); + let batchStarted!: () => void; + const batchStarting = new Promise(resolve => { batchStarted = resolve; }); + let releasePermission!: () => void; + const permissionReleased = new Promise(resolve => { releasePermission = resolve; }); + let permissionStarted!: () => void; + const permissionStarting = new Promise(resolve => { permissionStarted = resolve; }); + let reachedBoundary!: () => void; + const boundaryReached = new Promise(resolve => { reachedBoundary = resolve; }); + const textEvents = [1, 2, 3].map(seq => ({ + ...runnerEvent(seq, "item.delta", { kind: "agentMessage", text: `chunk-${seq}` }), priority: 1 as const, + })); + const permission = runnerEvent(4, "runtime_request.created", { request: { + schema: "paperclip.runtime_request.v2", requestKind: "permission_approval", requestId: "permission-1", + type: "permission", status: "pending", prompt: "Allow the proposed edit?", + choices: [{ key: "accept", label: "Allow once" }, { key: "cancel", label: "Cancel" }], + details: { toolCallId: "edit-1", kind: "edit" }, + origin: { adapter: "acpx-runtime-sidecar", provider: "acpx", method: "session/request_permission" }, + turnId: "turn-recovery", itemId: null, + } }); + const source = [...textEvents, permission, runnerEvent(5, "run.result.proposed", result), runnerEvent(6, "turn.completed")]; + let sourceReads = 0; + const session: NativeSession = { + identity: () => identity, + async capabilities() { return capabilities; }, + async *events() { + for (const event of source) { + sourceReads += 1; + if (event === permission) reachedBoundary(); + yield event; + } + }, + async startTurn() { return { turnId: "turn-recovery" }; }, + async result() { return { result, terminal, turnId: "turn-recovery" }; }, + async snapshot() { + return { backendKind: "mock", sessionId: "driver-recovery", identity, providerSessionId: "provider-recovery", + cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; + }, + async close() {}, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "blocked-authorization-fixture", version: "1", capabilities }; }, + async openSession() { return session; }, + }; + const committed: PrpEvent[] = []; + const makeReceipt = (event: PrpEvent) => ({ cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, + disposition: "committed" as const }); + const appendEvent = vi.fn(async event => { + const prp = event as PrpEvent; + if (prp === textEvents[0]) await firstReleased; + if (prp === permission) { permissionStarted(); await permissionReleased; } + committed.push(prp); + return makeReceipt(prp); + }); + const appendEvents = vi.fn>(async events => { + batchStarted(); + await batchReleased; + committed.push(...events); + return events.map(makeReceipt); + }); + const port: ControlPlanePort = { + async openRun() {}, async checkpointSession() {}, appendEvent, appendEvents, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + async completeRun() {}, + }; + const observed: PrpEvent[] = []; + const authorization = vi.fn((event: PrpEvent) => { + expect(committed.filter(entry => entry.sourceKind === "runner")).toEqual([...textEvents, permission]); + expect(event).toBe(permission); + }); + const execution = executeNativeSession({ input, backend, controlPlane: port, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resolveGovernedWait({ event }) { + observed.push(event); + if (event === permission) authorization(event); + return null; + } }); + void execution.catch(() => undefined); + try { + await boundaryReached; + await Promise.resolve(); + await Promise.resolve(); + expect(sourceReads).toBe(4); + expect(committed).toEqual([]); + expect(observed).toEqual([]); + expect(appendEvents).not.toHaveBeenCalled(); + expect(authorization).not.toHaveBeenCalled(); + + releaseFirst(); + await batchStarting; + expect(committed).toEqual([textEvents[0]]); + expect(observed).toEqual([textEvents[0]]); + expect(appendEvents).toHaveBeenCalledExactlyOnceWith(textEvents.slice(1), expect.objectContaining({ signal: expect.any(AbortSignal) })); + expect(appendEvent.mock.calls.map(([event]) => event)).toEqual([textEvents[0]]); + expect(sourceReads).toBe(4); + expect(authorization).not.toHaveBeenCalled(); + + releaseBatch(); + await permissionStarting; + expect(committed).toEqual(textEvents); + expect(observed).toEqual(textEvents); + expect(appendEvent).toHaveBeenLastCalledWith(permission, expect.objectContaining({ signal: expect.any(AbortSignal) })); + expect(sourceReads).toBe(4); + expect(authorization).not.toHaveBeenCalled(); + + releasePermission(); + await expect(execution).resolves.toMatchObject({ result, terminal }); + expect(authorization).toHaveBeenCalledExactlyOnceWith(permission); + expect(observed).toEqual([...textEvents, permission]); + expect(committed.filter(event => event.sourceKind === "runner")).toEqual(source); + expect(appendEvents.mock.calls.flatMap(([events]) => events)).toEqual(textEvents.slice(1)); + } finally { + releaseFirst(); releaseBatch(); releasePermission(); + await execution.catch(() => undefined); + } + }); + it("completes Cursor Cloud semantic planning through its default-only harness without losing the pinned revision", async () => { await writeFile(join(WORKSPACE, "cloud-plan-instructions.md"), "Make the plan only. Do not implement it.\n"); const prepared = preparedInput();