diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index cf0afdea076..2eafb2edc9b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2722,6 +2722,19 @@ is retained; complete token counters alone cannot certify a stopped provider. A timeout, missing accounting, or an unproven provider terminal remains a failure; the saved wait does not certify successful provider execution. +An exact governed-cancellation terminal may return its durable event receipt +before postcommit accounting and output observers finish. The receipt requires +the ordinary company, run, session, source and transaction fences; it is not a +physical-stop or accounting-completion receipt. The control plane retains the +observer debt and joins it before accepting a checkpoint or result. The existing +checkpoint deadline bounds that join, while the provider-terminal drain keeps its +existing deadline. Observer rejection preserves its original cause, and abort or +controller revocation prevents late checkpoint/result writes. A retained terminal +and saved wait can be replayed through duplicate-event observers after restart +without starting another model turn. Committed-only live events and trace +milestones are not re-emitted by that replay. Ordinary event appends still await +their observers. + Before either shutdown path exits, idle warm sessions close through their normal suspend-and-checkpoint path. Remote sessions therefore leave verified backup authority for the next controller even though their last run is already diff --git a/packages/paperclip-runner/src/contracts/control-plane-port.ts b/packages/paperclip-runner/src/contracts/control-plane-port.ts index b219c207b3a..b6e3aa685ea 100644 --- a/packages/paperclip-runner/src/contracts/control-plane-port.ts +++ b/packages/paperclip-runner/src/contracts/control-plane-port.ts @@ -28,6 +28,13 @@ export interface AppendControlPlaneEventOptions { * without committing before it rejects. */ signal: AbortSignal; + /** + * For an authoritative governed-cancellation terminal only, a port may + * return its committed receipt before postcommit observers finish. Such a + * port owns that observer debt and must join it before checkpoint or result + * acceptance. Ports that do not support deferral keep their usual behavior. + */ + deferPostCommitObservers?: true; } export interface CheckpointControlPlaneSessionOptions { diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 784f95ed6f4..df66c1fade0 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -4736,6 +4736,51 @@ describe("executeNativeSession recovery", () => { expect(lifecycle).toEqual(["closed"]); }); + it("settles an intentional governed park through the actual Codex session facade", async () => { + const transport = new FakeCodexTransport(); + const request = transport.request.bind(transport); + transport.request = async (method, params) => { + const response = await request(method, params); + if (method === "turn/interrupt") setImmediate(() => transport.push("turn/completed", { + threadId: "thread-1", turn: { id: "turn-1", status: "interrupted", error: null, items: [] }, + })); + return response; + }; + const executionInput = { ...input, workspace: { ...input.workspace, cwd: WORKSPACE } }; + const backend = createCodexNativeSessionBackend(executionInput, { + runnerInstanceId: "runner-recovery", transportFactory: () => transport, + }); + const events: PrpEvent[] = []; + const deferredEventTypes: string[] = []; + const completeRun = vi.fn(async () => undefined); + const execution = executeNativeSession({ input: executionInput, backend, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", timeoutMs: 5_000, + controlPlane: { + async openRun() {}, async checkpointSession() {}, completeRun, + async appendEvent(event, options) { + if (options?.deferPostCommitObservers) deferredEventTypes.push(event.eventType); + events.push(structuredClone(event)); return { + cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed", + }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + }, + resolveGovernedWait: ({ event }) => event.eventType === "item.completed" + && event.payload.kind === "agentMessage" ? yieldedResult : null, + }); + const outcome = expect(execution).resolves.toMatchObject({ result: yieldedResult, + terminal: { runTerminalState: "succeeded", reportedWorkDisposition: "yielded" } }); + await vi.waitFor(() => expect(transport.calls.some(call => call.method === "turn/start")).toBe(true)); + transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress" } }); + transport.push("item/completed", { threadId: "thread-1", turnId: "turn-1", + item: { id: "waiting-message", type: "agentMessage", channel: "final", text: "The human-only question is saved." } }); + await outcome; + expect(events.find(event => event.eventType === "turn.interrupted")?.payload).toMatchObject({ error: null, status: "interrupted" }); + expect(deferredEventTypes).toEqual(["turn.interrupted"]); + expect(events.some(event => event.eventType === "item.completed")).toBe(true); + expect(completeRun).toHaveBeenCalledOnce(); + expect(transport.calls.filter(call => call.method === "turn/start")).toHaveLength(1); + }); + it("commits a proved governed wait without waiting for abort-insensitive provider cleanup", async () => { vi.useFakeTimers(); try { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index a5f55fdf47a..3f0b6551bb2 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -1002,7 +1002,11 @@ async function consumeTurn( || (trailing.eventType === "item.completed" && trailing.payload.kind === "interrupt_acknowledgement") || (trailing.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(trailing.payload.runTerminalState))); if (trailing.turnId === event.turnId && (usage || terminal || cancellation)) { - const receipt = await controlPlane.appendEvent(trailing, { signal: drainAbort.signal }); + const receipt = await controlPlane.appendEvent(trailing, { + signal: drainAbort.signal, + ...(terminal && trailing.eventType !== "turn.failed" && trailing.payload.providerTerminalObserved !== false + ? { deferPostCommitObservers: true as const } : {}), + }); if (drainAbort.signal.aborted) return; eventCount += receipt.disposition === "committed" ? 1 : 0; highestContiguousSourceSeq = Math.max(highestContiguousSourceSeq, receipt.highestContiguousSourceSeq); diff --git a/server/src/__tests__/issue-monitor-scheduler.test.ts b/server/src/__tests__/issue-monitor-scheduler.test.ts index 1f6671e3ab2..3915b80fcad 100644 --- a/server/src/__tests__/issue-monitor-scheduler.test.ts +++ b/server/src/__tests__/issue-monitor-scheduler.test.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; import { costEvents, @@ -32,6 +32,21 @@ import { normalizeIssueExecutionPolicy, parseIssueExecutionState } from "../serv import { PaperclipRunnerToolAuthority } from "../services/native-runtime/paperclip-runner-tool-authority.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +// These process fixtures do not consume skills. Keep the real scheduler, +// execution lifecycle and selection snapshot, without a cold catalog expansion +// competing with the monitor's existing three-second settlement assertion. +vi.mock("../services/company-skills.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, companySkillService: (...args: Parameters) => { + const service = actual.companySkillService(...args); + return { ...service, withRuntimeSkillEntries: async (_companyId: string, + options: Parameters[1], + consume: (entries: Awaited>) => T | Promise) => { + await options.selectVersions?.([]); + return consume([]); + } }; + } }; +}); const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_MONITOR_TEST_DATABASE_URL"); const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 92aefdd9ad7..95d2b43cf45 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -64,7 +64,7 @@ import { assertCurrentWakeCommentsRead } from "./current-wake-comments.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; import { childReviewOutcomes } from "./child-review-outcomes.js"; import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js"; -import { materializeRuntimeQuestionFallback } from "./native-session-executor.js"; +import { materializeRuntimeQuestionFallback, nativeGovernedWaitResult } from "./native-session-executor.js"; import { subscribeAllCompanyLiveEvents, subscribeCompanyLiveEvents, @@ -314,6 +314,262 @@ describe("PaperclipControlPlanePort conformance", () => { } }); + it.each(["settles", "checkpoint timeout"] as const)("retains a durably recorded governed terminal while its postcommit observer is pending: %s", async mode => { + const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, issueId: randomUUID(), runId: randomUUID(), sessionId: randomUUID() }; + const runnerId = randomUUID(); + const fixtureContractId = randomUUID(); + await db.insert(issues).values({ id: identity.issueId, companyId: identity.companyId, title: "Human-only waiting checkpoint", + status: "in_progress", assigneeAgentId: identity.agentId }); + const [baseContract] = await db.select().from(completionContracts).where(eq(completionContracts.id, contractId)); + await db.insert(completionContracts).values({ ...baseContract!, id: fixtureContractId, issueId: identity.issueId }); + await db.insert(heartbeatRuns).values({ id: identity.runId, companyId: identity.companyId, agentId: identity.agentId, + status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, nativeSessionId: identity.sessionId, + runnerInstanceId: runnerId, completionContractId: fixtureContractId, completionContractSha256: contractSha }); + const interaction = await issueThreadInteractionService(db).create({ id: identity.issueId, companyId: identity.companyId }, { + kind: "ask_user_questions", title: "Choose the format", resolverPolicy: "human_only", continuationPolicy: "none", + sourceRunId: identity.runId, payload: { version: 1, title: "Choose the format", questions: [{ id: "format", + prompt: "Which format?", required: true, selectionMode: "single", options: [{ id: "onsite", label: "On site" }] }] }, + }, { agentId: identity.agentId, runId: identity.runId }); + let releaseObserver!: () => void; + const observerGate = new Promise(resolve => { releaseObserver = resolve; }); + let terminalWasRecorded = false; + let terminalRecorded!: () => void; + const recorded = new Promise(resolve => { terminalRecorded = resolve; }); + let observerFinished!: () => void; + const observerDone = new Promise(resolve => { observerFinished = resolve; }); + const port = new PaperclipControlPlanePort(db, { ...identity, + completionContractId: fixtureContractId, completionContractSha256: contractSha, + sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "park-control" }, { + async onCommittedEvent(event) { + if (event.eventType === "turn.interrupted") { terminalWasRecorded = true; terminalRecorded(); try { await observerGate; } finally { observerFinished(); } } + }, + }); + const contract = { revision: "phase6-v1", objective: "Conformance", criteria: [{ id: "objective", requirement: "Complete conformance" }] }; + const yielded = nativeGovernedWaitResult({ interaction: { id: interaction.id, + title: interaction.title ?? null, summary: interaction.summary ?? null }, completionContract: contract }); + const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record): PrpEvent => ({ + schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:${sourceSeq}`, sourceInstanceId: runnerId, sourceSeq, + sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "park-turn", + eventType, schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), payload, + }); + let starts = 0, terminalEmitted = false; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "park-observer", version: "1", + capabilities: { resume: false, typedEvents: true, steering: false, interruption: true, structuredResult: true } }; }, + async openSession() { return { identity: () => identity, + async capabilities() { return { resume: false, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async startTurn() { starts++; return { turnId: "park-turn" }; }, + cancel() { return { cleanup: Promise.resolve() }; }, + async *events() { yield event(1, "item.completed", { kind: "dynamicToolCall" }); terminalEmitted = true; yield event(2, "turn.interrupted", { status: "interrupted", error: null }); }, + async snapshot() { return { backendKind: "mock", sessionId: identity.sessionId, identity, + activeTurnId: terminalEmitted ? null : "park-turn", cursor: terminalEmitted ? "2" : "1" }; }, + async result() { return null; }, async close() {}, + }; }, + }; + const options = { backend, controlPlane: port, timeoutMs: 2_000, + ...(mode === "checkpoint timeout" ? { checkpointTimeoutMs: 100 } : {}), + runnerInstanceId: runnerId, controlPlaneInstanceId: "park-control", + input: parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v1", binding: { + companyId: identity.companyId, runId: identity.runId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "park-workspace" }, + task: { identifier: "PARK-1", title: "Wait for a human", description: null, prompt: "Wait for a human", workMode: "standard" }, + workspace: { cwd: process.cwd(), repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1 }, + provider: { kind: "codex", model: null }, interactionResponses: [], credentialBindings: [], + completionContract: { id: fixtureContractId, sha256: contractSha, schemaVersion: "paperclip.completion-contract.v1", contract } }), + resolveGovernedWait: ({ event }: { event: PrpEvent }) => event.eventType === "item.completed" ? yielded : null, + }; + const execution = executeNativeSession(options); + const outcome = execution.then(value => ({ value }), error => ({ error })); + try { + await Promise.race([recorded, outcome.then(result => { if ("error" in result) throw result.error; throw new Error("run settled before recording terminal"); })]); + const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId)); + expect(rows.some(row => row.eventType === "turn.interrupted")).toBe(true); + await new Promise(resolve => setTimeout(resolve, 700)); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]); + releaseObserver(); + await observerDone; + if (mode === "settles") expect(await outcome).not.toHaveProperty("error"); + else { + expect(await outcome).toMatchObject({ error: { message: "native session checkpoint timed out after 100ms" } }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]); + const restored = new PaperclipControlPlanePort(db, { ...identity, completionContractId: fixtureContractId, + completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "park-control" }); + const checkpoint = await restored.loadSessionCheckpoint(); + expect(checkpoint?.governedWait?.sourceEvent).toMatchObject({ sourceSeq: 1, turnId: "park-turn" }); + const session = await backend.openSession({ identity }); + const recovering = { ...backend, async recoverSession() { return { recovered: true as const, session }; } }; + await expect(executeNativeSession({ ...options, checkpointTimeoutMs: 1_000, controlPlane: restored, + backend: recovering, persistedSession: checkpoint })).resolves.toMatchObject({ result: yielded }); + } + expect(starts).toBe(1); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0]).toMatchObject({ + status: "pending", result: null, effectiveResolverPolicy: "human_only", continuationPolicy: "none", sourceRunId: identity.runId, + }); + } finally { releaseObserver(); await outcome; if (terminalWasRecorded) await observerDone; } + }); + + async function terminalObserverFixture(options: { + onCommittedEvent?: (event: PrpEvent) => Promise; + onDuplicateEvent?: (event: PrpEvent) => Promise; + assertControllerActive?: () => void; + } = {}) { + const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; + const runnerId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: identity.runId, companyId: identity.companyId, agentId: identity.agentId, + status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, nativeSessionId: identity.sessionId, + runnerInstanceId: runnerId, completionContractId: contractId, completionContractSha256: contractSha }); + const binding = { ...identity, completionContractId: contractId, completionContractSha256: contractSha, + sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "observer-control" }; + const port = new PaperclipControlPlanePort(db, binding, options); + await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId }); + const event: PrpEvent = { schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:1`, sourceInstanceId: runnerId, + sourceSeq: 1, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, + turnId: "observer-turn", eventType: "turn.interrupted", schemaVersion: 1, priority: 0, + emittedAt: new Date().toISOString(), payload: { status: "interrupted", error: null } }; + const snapshot = { backendKind: "mock" as const, sessionId: identity.sessionId, identity }; + const result = { result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }; + return { identity, runnerId, binding, port, event, snapshot, result }; + } + + it.each(["default", "deferred"] as const)("keeps %s terminal observation owned before checkpoint and result acceptance", async mode => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let observed = 0; + const f = await terminalObserverFixture({ async onCommittedEvent() { observed++; await gate; } }); + const signal = new AbortController().signal; + let appendSettled = false; + const append = f.port.appendEvent(f.event, { signal, ...(mode === "deferred" ? { deferPostCommitObservers: true } : {}) }) + .then(receipt => { appendSettled = true; return receipt; }); + let checkpointSettled = false, resultSettled = false; + try { + if (mode === "deferred") { + expect(await append).toMatchObject({ disposition: "committed" }); + const checkpoint = f.port.checkpointSession(f.snapshot, { signal }).then(() => { checkpointSettled = true; }); + const result = f.port.completeRun(f.result, { signal }).then(() => { resultSettled = true; }); + await new Promise(resolve => setTimeout(resolve, 30)); + expect(checkpointSettled).toBe(false); + expect(resultSettled).toBe(false); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined(); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]); + release(); await Promise.all([checkpoint, result]); + } else { + await new Promise(resolve => setTimeout(resolve, 30)); + expect(appendSettled).toBe(false); + } + } finally { release(); await append; } + expect(observed).toBe(1); + }); + + it("returns the default committed receipt when cancellation arrives during its postcommit observer", async () => { + let release!: () => void, observed!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { observed = resolve; }); + const f = await terminalObserverFixture({ async onCommittedEvent() { observed(); await gate; } }); + const controller = new AbortController(); + const outcome = f.port.appendEvent(f.event, { signal: controller.signal }).then(receipt => ({ receipt }), error => ({ error })); + try { + await started; + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.identity.runId))).toHaveLength(1); + const cancellation = new Error("cancelled after event transaction committed"); + controller.abort(cancellation); + release(); + expect(await outcome).toMatchObject({ receipt: { disposition: "committed" } }); + await expect(f.port.checkpointSession(f.snapshot, { signal: controller.signal })).rejects.toBe(cancellation); + await expect(f.port.completeRun(f.result, { signal: controller.signal })).rejects.toBe(cancellation); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]); + } finally { release(); await outcome; } + }); + + it("retains the original deferred observer rejection for checkpoint and result acceptance", async () => { + let reject!: (error: Error) => void; + const gate = new Promise((_resolve, no) => { reject = no; }); + const original = new Error("accounting persistence rejected"); + const f = await terminalObserverFixture({ async onCommittedEvent() { await gate; } }); + const signal = new AbortController().signal; + await f.port.appendEvent(f.event, { signal, deferPostCommitObservers: true }); + const checkpoint = f.port.checkpointSession(f.snapshot, { signal }).catch(error => error); + const result = f.port.completeRun(f.result, { signal }).catch(error => error); + reject(original); + expect(await checkpoint).toBe(original); expect(await result).toBe(original); + await expect(f.port.completeRun(f.result)).rejects.toBe(original); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined(); + }); + + it.each(["abort", "detach"] as const)("revokes a checkpoint and result waiting for deferred observation on %s", async reason => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let detached = false; + const original = new Error("controller authority revoked"); + const f = await terminalObserverFixture({ async onCommittedEvent() { await gate; }, + assertControllerActive() { if (detached) throw original; } }); + const controller = new AbortController(); + await f.port.appendEvent(f.event, { signal: controller.signal, deferPostCommitObservers: true }); + const checkpoint = f.port.checkpointSession(f.snapshot, { signal: controller.signal }).catch(error => error); + const result = f.port.completeRun(f.result, { signal: controller.signal }).catch(error => error); + if (reason === "abort") controller.abort(original); else detached = true; + release(); + expect(await checkpoint).toBe(original); expect(await result).toBe(original); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined(); + }); + + it("keeps a later observer error visible after an aborted checkpoint stops waiting", async () => { + let reject!: (error: Error) => void; + const gate = new Promise((_resolve, no) => { reject = no; }); + const f = await terminalObserverFixture({ async onCommittedEvent() { await gate; } }); + const controller = new AbortController(); + await f.port.appendEvent(f.event, { signal: controller.signal, deferPostCommitObservers: true }); + const checkpoint = f.port.checkpointSession(f.snapshot, { signal: controller.signal }).catch(error => error); + const stop = new Error("checkpoint authority revoked"); + controller.abort(stop); + expect(await checkpoint).toBe(stop); + const original = new Error("retained accounting failure after revocation"); + reject(original); + await expect(f.port.completeRun(f.result)).rejects.toBe(original); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined(); + }); + + it("joins the exact deferred observer on duplicate append and replays it after a new port opens", async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let observed = 0, replayed = 0; + const f = await terminalObserverFixture({ async onCommittedEvent() { observed++; await gate; }, + async onDuplicateEvent() { throw new Error("must join the original owned observation"); } }); + const signal = new AbortController().signal; + await f.port.appendEvent(f.event, { signal, deferPostCommitObservers: true }); + let duplicateSettled = false; + const duplicate = f.port.appendEvent(f.event, { signal }).then(value => { duplicateSettled = true; return value; }); + try { + await new Promise(resolve => setTimeout(resolve, 30)); expect(duplicateSettled).toBe(false); + release(); expect(await duplicate).toMatchObject({ disposition: "duplicate" }); + expect(observed).toBe(1); + const restored = new PaperclipControlPlanePort(db, f.binding, { async onDuplicateEvent() { replayed++; } }); + await restored.openRun({ identity: f.identity, backendKind: "mock", sourceInstanceId: f.runnerId }); + await expect(restored.appendEvent(f.event)).resolves.toMatchObject({ disposition: "duplicate" }); + expect(replayed).toBe(1); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.identity.runId))).toHaveLength(1); + } finally { release(); await duplicate; } + }); + + it.each(["company", "agent", "run", "session", "source", "failed", "unproven", "nonterminal"] as const)("does not grant deferred terminal authority to %s evidence", async fault => { + const f = await terminalObserverFixture(); + const event = { ...f.event, + ...(fault === "run" ? { runId: randomUUID() } : {}), + ...(fault === "session" ? { normalizedSessionId: randomUUID() } : {}), + ...(fault === "source" ? { sourceInstanceId: randomUUID() } : {}), + ...(fault === "failed" ? { eventType: "turn.failed" as const } : {}), + ...(fault === "unproven" ? { payload: { providerTerminalObserved: false } } : {}), + ...(fault === "nonterminal" ? { eventType: "item.completed" as const } : {}), + }; + const port = fault === "company" || fault === "agent" ? new PaperclipControlPlanePort(db, { ...f.binding, + ...(fault === "company" ? { companyId: randomUUID() } : { agentId: randomUUID() }) }) : f.port; + await expect(port.appendEvent(event, { signal: new AbortController().signal, deferPostCommitObservers: true })).rejects.toThrow(); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.identity.runId))).toEqual([]); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]); + }); + it("publishes native recovery metadata in the existing snapshot transaction and fences stale generations", async () => { const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; const runnerId = randomUUID(); @@ -406,6 +662,7 @@ describe("PaperclipControlPlanePort conformance", () => { it.each([ ["checkpoint", "detach"], ["result", "detach"], ["event", "detach"], ["checkpoint", "abort"], ["result", "abort"], ["event", "abort"], + ["deferred event", "detach"], ["deferred event", "abort"], ] as const)("revokes an in-flight %s on %s", async (operation, reason) => { const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; const runnerId = randomUUID(); @@ -424,12 +681,13 @@ describe("PaperclipControlPlanePort conformance", () => { assertControllerActive: () => { if (detached) throw new Error("controller_detached"); }, }); await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId }); - const mutate = () => operation === "event" ? port.appendEvent({ + const mutate = () => operation === "event" || operation === "deferred event" ? port.appendEvent({ schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:1`, sourceSeq: 1, sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, eventType: "turn.interrupted", schemaVersion: 1, + turnId: "revoked-turn", priority: 0, emittedAt: new Date().toISOString(), payload: { reason: "governed_wait" }, - }, { signal: controller.signal }) : operation === "checkpoint" + }, { signal: controller.signal, ...(operation === "deferred event" ? { deferPostCommitObservers: true } : {}) }) : operation === "checkpoint" ? port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity }, { signal: controller.signal }) : port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }, { signal: controller.signal }); let pending: Promise | undefined; diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index 4e85a8ef843..ac0f582c445 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -98,6 +98,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort { readonly #nativeRecovery?: { compatibility: NativeControllerCompatibility; owner: NativeRunnerCheckpointOwner }; readonly #taskSessionPublication?: NativeTaskSessionPublicationContext; readonly #taskSessionPublicationOwner?: NativeRunnerCheckpointOwner; + readonly #deferredEventObservers = new Map>(); + #deferredObserverFailure: { error: unknown } | null = null; constructor( db: Db, @@ -137,6 +139,33 @@ export class PaperclipControlPlanePort implements ControlPlanePort { options?.signal.throwIfAborted(); } + async #settleDeferredObservers(options?: { signal: AbortSignal }): Promise { + const assertActive = () => { + this.#assertActive(options); + if (this.#deferredObserverFailure) throw this.#deferredObserverFailure.error; + }; + assertActive(); + if (this.#deferredEventObservers.size === 0) return; + const signal = options?.signal; + let removeAbort = () => {}; + const aborted = new Promise((_resolve, reject) => { + if (!signal) return; + const abort = () => reject(signal.reason ?? new Error("native event observation aborted")); + signal.addEventListener("abort", abort, { once: true }); + removeAbort = () => signal.removeEventListener("abort", abort); + if (signal.aborted) abort(); + }); + try { + // Aborting a checkpoint revokes its writes, not already-owned accounting + // or output observation. Each retained promise remains observed below. + while (this.#deferredEventObservers.size > 0) { + await Promise.race([Promise.all([...this.#deferredEventObservers.values()]), aborted]); + assertActive(); + } + } finally { removeAbort(); } + assertActive(); + } + #matchesPersistedBinding(run: Pick): boolean { @@ -207,6 +236,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { async checkpointSession(snapshot: PersistedNativeSession, options?: { signal: AbortSignal }): Promise { this.#assertActive(options); + if (this.#deferredEventObservers.size > 0 || this.#deferredObserverFailure) await this.#settleDeferredObservers(options); snapshot = redactRunnerApiPayloads(this.#identityRedactor.redact(snapshot)); const identity = snapshot.identity; if ( @@ -324,7 +354,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { }); } - async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal }) { + async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal; deferPostCommitObservers?: true }) { const persistenceStartedAt = Date.now(), persistenceClock = performance.now(); this.#assertActive(options); if (!isPrpEvent(value)) throw new Error("native_legacy_event_not_supported"); @@ -340,6 +370,10 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (event.sourceInstanceId !== expectedSourceInstanceId) { throw new Error("native_event_source_binding_mismatch"); } + if (options?.deferPostCommitObservers && ( + !["turn.completed", "turn.interrupted", "turn.cancelled"].includes(event.eventType) + || !event.turnId || event.payload.providerTerminalObserved === false + )) throw new Error("native_deferred_observer_terminal_required"); const terminalItem = /^item\.(completed|failed|cancelled)$/.test(event.eventType); const terminalTurn = /^(turn|session)\.(completed|failed|cancelled|interrupted|closed)$/.test(event.eventType); if ((event.eventType === "item.delta" || terminalItem || terminalTurn) && this.#identityRedactor.values.length > 0) { @@ -396,32 +430,48 @@ export class PaperclipControlPlanePort implements ControlPlanePort { return receipt; }); this.#assertActive(options); - try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceStartedAt + performance.now() - persistenceClock); } catch {} - if (persisted.disposition === "committed") { - try { - publishLiveEvent({ - companyId: this.#binding.companyId, - type: "heartbeat.run.event", - payload: { - runId: this.#binding.runId, - agentId: this.#binding.agentId, - issueId: this.#binding.issueId, - seq: persisted.row.seq, - eventType: persisted.row.eventType, - stream: persisted.row.stream, - level: persisted.row.level, - payload: persisted.row.payload, - }, - }); - } catch { + const persistenceEndedAt = persistenceStartedAt + performance.now() - persistenceClock; + const observerKey = JSON.stringify([event.sourceInstanceId, event.sourceEventId]); + let observation = this.#deferredEventObservers.get(observerKey); + if (!observation) observation = (async () => { + try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceEndedAt); } catch {} + if (persisted.disposition === "committed") { + try { + publishLiveEvent({ + companyId: this.#binding.companyId, + type: "heartbeat.run.event", + payload: { + runId: this.#binding.runId, + agentId: this.#binding.agentId, + issueId: this.#binding.issueId, + seq: persisted.row.seq, + eventType: persisted.row.eventType, + stream: persisted.row.stream, + level: persisted.row.level, + payload: persisted.row.payload, + }, + }); + } catch { + } + await this.#onCommittedEvent?.(event, persisted.row.seq); + } else { + // A recovered runner may replay the event whose durable side effects + // parked the prior attempt. Do not repeat those effects, but let the + // embedding runtime refresh observational state before appendEvent + // returns to its synchronous governed-wait boundary. + await this.#onDuplicateEvent?.(event, persisted.row.seq); } - await this.#onCommittedEvent?.(event, persisted.row.seq); + })(); + if (options?.deferPostCommitObservers) { + this.#deferredEventObservers.set(observerKey, observation); + void observation.then(() => { + if (this.#deferredEventObservers.get(observerKey) === observation) this.#deferredEventObservers.delete(observerKey); + }, error => { + this.#deferredObserverFailure ??= { error }; + if (this.#deferredEventObservers.get(observerKey) === observation) this.#deferredEventObservers.delete(observerKey); + }); } else { - // A recovered runner may replay the event whose durable side effects - // parked the prior attempt. Do not repeat those effects, but let the - // embedding runtime refresh observational state before appendEvent - // returns to its synchronous governed-wait boundary. - await this.#onDuplicateEvent?.(event, persisted.row.seq); + await observation; } return { cursor: persisted.row.seq, @@ -462,6 +512,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput, options?: { signal: AbortSignal }): Promise { this.#assertActive(options); + if (this.#deferredEventObservers.size > 0 || this.#deferredObserverFailure) await this.#settleDeferredObservers(options); value = redactRunnerApiPayloads(this.#identityRedactor.redact(value)); if (!isCompleteInput(value)) throw new Error("native_structured_result_required"); // Capture compatibility diagnostics before canonical validation removes