From 386433f178ed564f1da78e20873437bf8600e0a1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 9 Aug 2026 04:19:01 +0000 Subject: [PATCH] fix(native-runner): complete recovery and conformance Co-Authored-By: Paperclip --- .../design/phase-6-thin-paperclip-adapter.md | 16 +- .../docs/phase-06-thin-paperclip-adapter.md | 30 ++- .../phase-06-thin-paperclip-adapter.md | 15 +- .../2026-08-09-phase-06-verification.md | 43 ++-- .../2026-08-09-phase-06-implementation.md | 29 ++- .../conformance/control-plane-port.test.ts | 5 + .../src/conformance/control-plane-port.ts | 63 +++++ .../src/native-session-runtime.test.ts | 92 ++++++++ .../src/native-session-runtime.ts | 21 +- .../legacy-finalization-regression.test.ts | 133 ++++++++--- .../native-finalization-migration.test.ts | 121 +++++++--- .../native-interaction-bridge.test.ts | 172 ++++++++++++++ .../native-session-resumption.test.ts | 184 +++++++++++++++ .../native-status-arbiter-corpus.test.ts | 216 +++++++++++++----- server/src/services/heartbeat.ts | 130 +++++++++-- server/src/services/native-runtime/index.ts | 1 + .../native-runtime/native-execution-input.ts | 9 +- .../native-finalization-reconciler.ts | 152 +++++++++++- .../native-interaction-bridge.ts | 128 ++++++++++- .../native-session-executor.test.ts | 27 ++- .../native-runtime/native-session-executor.ts | 85 +++++-- .../paperclip-control-plane-port.test.ts | 5 + 22 files changed, 1454 insertions(+), 223 deletions(-) create mode 100644 server/src/__tests__/native-interaction-bridge.test.ts create mode 100644 server/src/__tests__/native-session-resumption.test.ts diff --git a/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md b/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md index b457e2d0462..76588ec43dc 100644 --- a/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md @@ -703,16 +703,30 @@ from the provisional list above. They are added without expanding authority: ```text packages/paperclip-runner/src/native-session-runtime.ts packages/paperclip-runner/src/native-session-runtime.test.ts +packages/paperclip-runner/src/contracts/native-execution.ts +packages/paperclip-runner/src/contracts/native-execution.test.ts +packages/paperclip-runner/src/backends/codex-native-backend.ts packages/paperclip-runner/spec/fixtures/status-authority-phase5.json +packages/adapter-utils/package.json +packages/shared/src/index.ts +packages/shared/src/types/index.ts +packages/shared/src/validators/index.ts server/src/services/native-runtime/native-session-executor.ts server/src/services/native-runtime/native-session-executor.test.ts server/src/services/native-runtime/evidence-classifier.test.ts server/src/services/native-runtime/paperclip-control-plane-port.test.ts server/src/services/native-runtime/status-arbiter.test.ts +server/src/services/recovery/service.ts packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md ``` -The package runtime files own persisted provider-session recovery. The corpus +The package runtime, closed native-input contract, and Codex backend files own +persisted provider-session recovery and construction. The adapter-utils +metadata declares the package dependency used by the approved native +finalization result type; the shared barrels only expose the already-reviewed +types and validators. The recovery service touch reuses the existing +source-scoped recovery action and wake path rather than creating new authority. +The corpus edit moves two existing coverage labels between fixtures without changing any fixture input or expected outcome, eliminating an unjoinable fixture. The server executor files own only the run-scoped cancellation handle and coordinator diff --git a/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md b/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md index 5d9eed2df22..3ab7a26972d 100644 --- a/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md @@ -40,8 +40,9 @@ contract are persisted before provider execution. - `completeRun` validates and idempotently persists the immutable structured result plus terminal fact. -The package-owned `HarnessDriverBackend` is the approved -`NativeSessionBackend`. Core passes only the closed `NativeExecutionInputV1`. +The package-owned Codex native backend is the production +`NativeSessionBackend`; the harness backend remains the deterministic test +adapter. Core passes only the closed `NativeExecutionInputV1`. The model receives the smaller `NativeModelEnvelopeV1`; bindings and credential references do not cross that boundary. @@ -64,14 +65,23 @@ status-version CAS and audit record. ## Recovery and safety -The finalization reconciler selects persisted `runtime_mode=native` rows; it -does not consult the current flag. Per-attempt leases prevent concurrent -session/finalization ownership, while retryable failures retain the immutable -result and a first-class recovery action. A recovered session replays its -checkpoint and appends only missing control-plane facts instead of opening a -second provider session. Cancellation uses a run-scoped normalized session -handle before the existing process cleanup. Native execution does not construct -a Paperclip JWT, managed MCP access, legacy context, or raw adapter environment. +Recovery selects persisted `runtime_mode=native` coordinators and does not +consult the current flag. Result-less transport loss retains the original run, +closed native input, provider checkpoint, retry attempt, and recovery action; +a database lease dispatches that same run after restart. Missing checkpoints, +cancellation, and retry exhaustion fail closed without opening a new provider +session or falling into legacy. Result-bearing rows continue through the +workspace/finalization reconciler. A recovered package session inspects the +provider snapshot before starting a turn, so an active provider turn is not +duplicated. Cancellation uses a run-scoped normalized session handle before the +existing process cleanup. Native execution does not construct a Paperclip JWT, +managed MCP access, legacy context, or raw adapter environment. + +Resolved `request_confirmation` and `ask_user_questions` wakes are projected +through the existing authorized issue-interaction service and bound to the +company, issue, run, agent, and interaction identities in the persisted input. +Governed tool actions, unsupported kinds, unresolved rows, and self-resolution +fail closed; no credentials enter the model envelope. ## Verification status diff --git a/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md b/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md index 760a647de63..deb3abc0ec9 100644 --- a/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md @@ -51,21 +51,26 @@ pnpm --filter @paperclipai/server exec vitest run \ src/__tests__/native-runner-input-boundary.test.ts \ src/__tests__/native-run-finalizer.test.ts \ src/__tests__/native-status-arbiter-corpus.test.ts \ + src/__tests__/native-interaction-bridge.test.ts \ + src/__tests__/native-session-resumption.test.ts \ src/__tests__/native-finalization-recovery.test.ts \ src/__tests__/native-finalization-migration.test.ts \ - src/__tests__/legacy-finalization-regression.test.ts + src/__tests__/legacy-finalization-regression.test.ts \ + src/services/native-runtime/native-session-executor.test.ts ``` This runs the unchanged package conformance suite against the real Paperclip -port. Twenty-eight targeted tests check company binding, duplicate/gap replay, +port. Thirty-nine targeted tests check company binding, duplicate/gap replay, checkpoint recovery, immutable result ingestion, durable-evidence authority, workspace-gated finalization, atomic liveness rollback, status-version CAS, -migration repair, legacy byte equivalence, and audit output. All database tests +migration repair, legacy byte equivalence, typed interaction materialization, +same-run pre-result resumption, lease races, and audit output. All database tests start embedded PostgreSQL and do not skip when a developer database URL is absent. -The command above is the repeatable internal canary and post-kill-switch legacy -proof. The remaining sections are an optional live-provider rollout procedure; +The command above is the repeatable scripted internal canary and actual +flag-off legacy-heartbeat proof. It is not a live-provider canary. The remaining +sections are an optional live-provider rollout procedure; they were not run during remediation and require explicit operator authority to change an instance flag and agent profile. diff --git a/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md b/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md index 80580eba117..16e2c4a141e 100644 --- a/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md +++ b/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md @@ -30,7 +30,7 @@ pnpm --filter @paperclipai/paperclip-runner trace:phase6 -- \ --target mock --scenario happy-path ``` -Result: four files and six tests passed. The stable trace reported native +Result: four files and seven tests passed. The stable trace reported native mode, `runStatus=succeeded`, three events, contiguous source sequence 3, `workspaceFinalizeStatus=succeeded`, zero legacy invocations, and a server-owned `done` decision with status version 0 -> 1. Restart recovery reused the bound @@ -39,7 +39,7 @@ existing accepted result event, and appended only the missing terminal fact. # Real Paperclip adapter and public-session task -The complete acceptance gate below passed 28 tests in ten files against +The complete acceptance gate below passed 39 tests in thirteen files against embedded PostgreSQL with zero skipped tests. Database-backed tests start their own database instead of skipping when a developer `DATABASE_URL` is absent. The same package conformance suite passed unchanged against @@ -59,7 +59,14 @@ The focused corpus also proved: - remote, non-standard, or non-Codex explicit profiles fail closed; - company/run/agent/issue/source bindings reject mismatches; - source gaps remain visible, identical retries deduplicate, and replay is - exclusive and byte-stable; + exclusive and byte-stable; mutated identities, sequences, results, and + bindings fail closed in the shared mock/real port suite; +- result-less transport loss dispatches the original persisted native run from + its closed envelope/provider checkpoint after a database lease, including + flag-off restart and active-turn recovery without a second session or turn; +- resolved confirmations and question answers enter the persisted native input + only through the authorized issue-interaction service; governed, unresolved, + unsupported, or self-approved paths fail closed without credentials; - incomplete/review/yield/cancelled results receive a durable review, continuation, or recovery path; - only a task-wide blocker with a named owner and action selects `blocked`; @@ -72,7 +79,7 @@ The focused corpus also proved: bindings, recovery, status, decisions, and effects roll back together before retry ownership is recorded. -The acceptance-matrix entry points also passed as one 10-file, 28-test gate: +The acceptance-matrix entry points passed as one 13-file, 39-test gate: ```sh pnpm --filter @paperclipai/server exec vitest run \ @@ -82,16 +89,22 @@ pnpm --filter @paperclipai/server exec vitest run \ src/__tests__/native-runner-input-boundary.test.ts \ src/__tests__/native-runner-phase6.integration.test.ts \ src/__tests__/native-finalization-recovery.test.ts \ + src/__tests__/native-interaction-bridge.test.ts \ + src/__tests__/native-session-resumption.test.ts \ src/__tests__/heartbeat-run-event-sequencing.test.ts \ src/__tests__/legacy-finalization-regression.test.ts \ src/__tests__/native-finalization-migration.test.ts \ - src/__tests__/native-status-arbiter-corpus.test.ts + src/__tests__/native-status-arbiter-corpus.test.ts \ + src/services/native-runtime/native-session-executor.test.ts ``` This gate proves concurrent event allocation, duplicate-only migration repair, -bounded retry exhaustion, kill-switch legacy byte compatibility, and complete -Section 18.13 fixture-to-consumer traceability in addition to the direct suites. -It is the repeatable internal native canary and post-kill-switch legacy proof. +bounded retry exhaustion, actual flag-off legacy execution, and executable +Section 18.13 fixture-to-consumer traceability. Each of the 52 fixtures and 70 +unique matrix rows invokes production consumers; the test hashes observed +decisions/effects, never the fixture's `expected` object. The selected-task +canary remains scripted at the provider boundary, while the legacy heartbeat is +an actual adapter/finalizer execution. # Compile and migration checks @@ -118,14 +131,17 @@ pnpm --filter @paperclipai/server exec vitest run \ src/__tests__/heartbeat-run-event-sequencing.test.ts ``` -Result: five files and 11 tests passed with zero skips. The production-shaped -0211 rehearsal began with legacy event sequences `[1,5,5,9]`, retained the +Result: five files and 11 tests passed with zero skips. The 0211 rehearsal +reconstructed a production-shaped pre-0211 schema, removed the 0211 journal +entry, seeded legacy event sequences `[1,5,5,9]`, and applied the complete +migration. It retained the original unique cursors and first `5`, moved only the later duplicate to `10`, seeded `next_event_seq=11`, preserved every non-sequence field byte-for-byte, and enforced uniqueness. Thirty-two concurrent mixed event writers produced a gap-free allocator stream. Invalid native finalization exhausted its -three-attempt budget into named recovery without touching issue status. The -legacy snapshot remained byte-equivalent with zero native history rows. +three-attempt budget into named recovery without touching issue status. An +actual flag-off heartbeat executed the legacy adapter/finalizer, remained +byte-equivalent across native reconciliation, and created zero native rows. # Credential, governance, budget, and legacy boundaries @@ -133,7 +149,8 @@ legacy snapshot remained byte-equivalent with zero native history rows. proves the model envelope omits company/run/agent bindings and credential references. The native branch never creates a local agent JWT, MCP gateway, raw adapter environment, or legacy context. Runtime requests cannot auto-approve; -the native interaction bridge rejects unsupported requests explicitly. +the native interaction bridge materializes supported typed responses through +the authorized service and rejects unsupported/governed/self-approved paths. Explicit cancellation, agent pause, and budget hard-stop share `cancelRunInternal`, which now invokes the run-scoped normalized native session diff --git a/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md b/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md index f0a3fd6f149..48eae4893a4 100644 --- a/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md +++ b/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md @@ -40,21 +40,30 @@ existing server authority and keep legacy execution as the default. 10. Treat model citations as claims. Only server-verifiable durable records can satisfy completion criteria, and every non-terminal decision must create or bind a real review, wake, blocker, or recovery path in the status transaction. +11. Persist the closed native input before opening the provider. Result-less + transport failures keep the same run live behind a bounded database lease; + restart, flag disablement, and an already-active provider turn cannot create + a second provider session, run, or turn. +12. Project supported resolved interaction responses through the existing + authorized issue-interaction service. Governed, unresolved, unsupported, + cross-boundary, or self-approved requests fail closed without credentials. # Evidence - [Phase 6 verification](../evidence/2026-08-09-phase-06-verification.md) - [Phase 6 reference](../../docs/phase-06-thin-paperclip-adapter.md) - [Runnable tutorial](../../docs/tutorials/phase-06-thin-paperclip-adapter.md) -- Package conformance and recovery: four files, six tests passed. -- Phase 6 acceptance-matrix entry points: ten files, twenty-eight tests passed +- Package conformance and recovery: four files, seven tests passed. +- Phase 6 acceptance-matrix entry points: thirteen files, thirty-nine tests passed with zero skips, including the database-backed selected-task canary, six atomic-liveness failpoints, migration, sequencing, bounded recovery, and legacy compatibility. -- Focused database rehearsal: five files, eleven tests passed with zero skips; - the legacy event/read snapshot was byte-equivalent and had zero native rows. -- Section 18.13 source corpus: 52 fixtures, all 70 matrix rows, and 18 structural - checks passed with every fixture bound to a named consumer. +- The migration rehearsal rewound to a production-shaped pre-0211 schema and + applied the complete migration; the actual flag-off heartbeat executed the + legacy adapter/finalizer and created zero native rows. +- Section 18.13 source corpus: 52 fixture executions and 70 unique matrix-row + executions passed through production consumers with observed result/effect + digests; expected-object hashing and import-only aliases were removed. - Runner, shared, server, and database typechecks passed; migration safety passed. @@ -75,14 +84,16 @@ existing server authority and keep legacy execution as the default. - Phase 6 supports local, standard, workspace-backed `codex_local` tasks only. - It adds no browser UI or public runner WebSocket. -- Typed provider requests are rejected unless an existing Paperclip governance - path can materialize an authorized response; there is no auto-approval. +- Resolved confirmations and question answers are materialized only through the + existing authorized Paperclip interaction service; governed and unsupported + requests cannot auto-approve. - Live operator proof depends on a board-authorized local instance and an already authenticated Codex installation; deterministic CI uses the same public session contract with a scripted backend. - The remediation did not dispatch a new live provider task or mutate rollout settings. The verified “internal canary” and post-kill-switch trace are the - real embedded-PostgreSQL selected-task and legacy-snapshot cases; the live + embedded-PostgreSQL scripted selected-task and actual flag-off legacy + heartbeat cases; the live tutorial remains an explicit operator checkpoint. # Follow-up questions diff --git a/packages/paperclip-runner/src/conformance/control-plane-port.test.ts b/packages/paperclip-runner/src/conformance/control-plane-port.test.ts index 87237a12068..ceea2e80503 100644 --- a/packages/paperclip-runner/src/conformance/control-plane-port.test.ts +++ b/packages/paperclip-runner/src/conformance/control-plane-port.test.ts @@ -19,6 +19,11 @@ describe("ControlPlanePort conformance", () => { highestContiguousSourceSeq: 3, duplicateDisposition: "duplicate", terminalReplayIdempotent: true, + openBindingRejected: true, + eventIdMutationRejected: true, + eventSequenceMutationRejected: true, + replayBindingRejected: true, + resultMutationRejected: true, }); }); diff --git a/packages/paperclip-runner/src/conformance/control-plane-port.ts b/packages/paperclip-runner/src/conformance/control-plane-port.ts index 92f83ff5b31..767e0c37252 100644 --- a/packages/paperclip-runner/src/conformance/control-plane-port.ts +++ b/packages/paperclip-runner/src/conformance/control-plane-port.ts @@ -10,6 +10,11 @@ export interface ControlPlanePortConformanceSnapshot { highestContiguousSourceSeq: number; duplicateDisposition: "duplicate"; terminalReplayIdempotent: boolean; + openBindingRejected: boolean; + eventIdMutationRejected: boolean; + eventSequenceMutationRejected: boolean; + replayBindingRejected: boolean; + resultMutationRejected: boolean; } export interface ControlPlanePortConformanceHarness { @@ -83,6 +88,16 @@ export async function runControlPlanePortConformance( await harness.start?.(); try { await harness.port.openRun(CONTROL_PLANE_CONFORMANCE_OPEN); + let openBindingRejected = false; + try { + await harness.port.openRun({ + ...CONTROL_PLANE_CONFORMANCE_OPEN, + identity: { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, issueId: "forged-issue" }, + }); + } catch { + openBindingRejected = true; + } + if (!openBindingRejected) throw new Error("mismatched run binding was accepted"); const first = await harness.port.appendEvent(CONTROL_PLANE_CONFORMANCE_EVENTS[0]); if (first.disposition !== "committed" || first.highestContiguousSourceSeq !== 1) { throw new Error("first event did not commit at cursor 1"); @@ -93,6 +108,27 @@ export async function runControlPlanePortConformance( if (recovered.highestContiguousSourceSeq !== 3) throw new Error("source gap did not recover"); const duplicate = await harness.port.appendEvent(CONTROL_PLANE_CONFORMANCE_EVENTS[1]); if (duplicate.disposition !== "duplicate") throw new Error("identical duplicate was not idempotent"); + let eventIdMutationRejected = false; + try { + await harness.port.appendEvent({ + ...CONTROL_PLANE_CONFORMANCE_EVENTS[1], + payload: { mutated: true }, + }); + } catch { + eventIdMutationRejected = true; + } + if (!eventIdMutationRejected) throw new Error("mutated event id replay was accepted"); + let eventSequenceMutationRejected = false; + try { + await harness.port.appendEvent({ + ...CONTROL_PLANE_CONFORMANCE_EVENTS[1], + sourceEventId: "runner-phase6-conformance:mutated-sequence-two", + payload: { mutated: true }, + }); + } catch { + eventSequenceMutationRejected = true; + } + if (!eventSequenceMutationRejected) throw new Error("mutated source sequence replay was accepted"); const replay = await harness.port.replayEvents({ runId: CONTROL_PLANE_CONFORMANCE_OPEN.identity.runId, @@ -103,6 +139,18 @@ export async function runControlPlanePortConformance( if (replay.events.map((entry) => entry.sourceSeq).join(",") !== "2,3") { throw new Error("exclusive replay cursor returned the wrong event set"); } + let replayBindingRejected = false; + try { + await harness.port.replayEvents({ + runId: "forged-run", + sourceInstanceId: "runner-phase6-conformance", + afterSourceSeq: 0, + limit: 10, + }); + } catch { + replayBindingRejected = true; + } + if (!replayBindingRejected) throw new Error("mismatched replay binding was accepted"); const completion: CompleteControlPlaneRunInput = { result: CONTROL_PLANE_CONFORMANCE_RESULT, @@ -113,11 +161,26 @@ export async function runControlPlanePortConformance( }; await harness.port.completeRun(completion); await harness.port.completeRun(completion); + let resultMutationRejected = false; + try { + await harness.port.completeRun({ + ...completion, + result: { ...completion.result, summary: "mutated result replay" }, + }); + } catch { + resultMutationRejected = true; + } + if (!resultMutationRejected) throw new Error("mutated result replay was accepted"); return { eventCount: replay.highestContiguousSourceSeq, highestContiguousSourceSeq: replay.highestContiguousSourceSeq, duplicateDisposition: duplicate.disposition, terminalReplayIdempotent: true, + openBindingRejected, + eventIdMutationRejected, + eventSequenceMutationRejected, + replayBindingRejected, + resultMutationRejected, }; } finally { await harness.stop?.(); diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 42f7165a42a..abcf58e4149 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -101,6 +101,98 @@ function highestContiguous(events: PrpEvent[]): number { } describe("executeNativeSession recovery", () => { + it("continues a provider-reported active turn without starting a duplicate turn", async () => { + const checkpoint: PersistedNativeSession = { + backendKind: "mock", + sessionId: "driver-recovery", + identity, + providerSessionId: "provider-recovery", + cursor: "0", + activeTurnId: null, + pendingRuntimeRequests: [], + lineage: [], + }; + const providerSnapshot: PersistedNativeSession = { + ...checkpoint, + cursor: "1", + activeTurnId: "turn-recovery", + }; + const terminalEvent: PrpEvent = { + schema: "paperclip.prp.event.v1", + sourceEventId: "provider-recovery:1", + sourceSeq: 1, + sourceInstanceId: "provider-recovery", + sourceKind: "provider", + runId: identity.runId, + normalizedSessionId: identity.sessionId, + turnId: "turn-recovery", + eventType: "turn.completed", + schemaVersion: 1, + priority: 0, + emittedAt: "2026-08-09T00:00:00.000Z", + payload: {}, + }; + const bySource = new Map(); + const startTurn = vi.fn(async () => ({ turnId: "duplicate-turn" })); + const openSession = vi.fn(async () => { throw new Error("must recover the provider session"); }); + const session: NativeSession = { + identity: () => identity, + async capabilities() { + return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; + }, + async *events() { yield terminalEvent; }, + startTurn, + async result() { return { result, terminal, turnId: "turn-recovery" }; }, + async snapshot() { return structuredClone(providerSnapshot); }, + async close() {}, + }; + const backend: NativeSessionBackend = { + async descriptor() { + return { + kind: "mock", + name: "recovery-backend", + version: "1", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + }; + }, + openSession, + async recoverSession() { return { recovered: true, session }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, + async loadSessionCheckpoint() { return structuredClone(checkpoint); }, + async checkpointSession() {}, + async appendEvent(event) { + const list = bySource.get(event.sourceInstanceId) ?? []; + list.push(structuredClone(event)); + bySource.set(event.sourceInstanceId, list); + return { + cursor: list.length, + highestContiguousSourceSeq: highestContiguous(list), + disposition: "committed", + }; + }, + async replayEvents(replay) { + const list = bySource.get(replay.sourceInstanceId) ?? []; + return { + events: structuredClone(list.filter((event) => event.sourceSeq > replay.afterSourceSeq)), + highestContiguousSourceSeq: highestContiguous(list), + }; + }, + async completeRun() {}, + }; + + await expect(executeNativeSession({ + input, + backend, + controlPlane: port, + runnerInstanceId: "runner-recovery", + controlPlaneInstanceId: "control-recovery", + })).resolves.toMatchObject({ turnId: "turn-recovery", providerSessionId: "provider-recovery" }); + expect(openSession).not.toHaveBeenCalled(); + expect(startTurn).not.toHaveBeenCalled(); + }); + it("recovers a completed checkpoint and appends only a missing control terminal fact", async () => { const checkpoint: PersistedNativeSession = { backendKind: "mock", diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 2c17061e50e..dfbac0e2acf 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -120,19 +120,24 @@ export async function executeNativeSession(options: ExecuteNativeSessionOptions) await options.controlPlane.checkpointSession(await session.snapshot()); } }; - await checkpoint(); + const recoveredSnapshot = await session.snapshot(); + await options.controlPlane.checkpointSession?.(recoveredSnapshot); let consumed = { event: null as PrpEvent | null, eventCount: 0, highestContiguousSourceSeq: 0 }; - let completed = persistedSession?.semanticResult && persistedSession.terminal + const completionSnapshot = recoveredSnapshot.semanticResult && recoveredSnapshot.terminal + ? recoveredSnapshot + : persistedSession; + let completed = completionSnapshot?.semanticResult && completionSnapshot.terminal ? { - result: persistedSession.semanticResult, - terminal: persistedSession.terminal, - turnId: persistedSession.activeTurnId ?? null, + result: completionSnapshot.semanticResult, + terminal: completionSnapshot.terminal, + turnId: completionSnapshot.activeTurnId ?? null, } : null; if (!completed) { const consuming = consumeTurn(session, options.controlPlane, options.timeoutMs ?? 900_000); - if (!recovered || !persistedSession?.activeTurnId) { + const recoveredActiveTurnId = recoveredSnapshot.activeTurnId ?? persistedSession?.activeTurnId ?? null; + if (!recovered || !recoveredActiveTurnId) { await session.startTurn({ message: { role: "user", text: JSON.stringify(buildNativeModelEnvelope(input)) }, }); @@ -145,8 +150,8 @@ export async function executeNativeSession(options: ExecuteNativeSessionOptions) } if (completed === null) throw new Error("native_finalization_missing: session returned no semantic result"); let terminal: PrpTerminalState; - if (persistedSession?.semanticResult && persistedSession.terminal) { - terminal = persistedSession.terminal; + if (completionSnapshot?.semanticResult && completionSnapshot.terminal) { + terminal = completionSnapshot.terminal; } else { terminal = terminalFromEvent(consumed.event!, completed.result.reportedWorkDisposition); } diff --git a/server/src/__tests__/legacy-finalization-regression.test.ts b/server/src/__tests__/legacy-finalization-regression.test.ts index 275eeb024da..fd440c26a56 100644 --- a/server/src/__tests__/legacy-finalization-regression.test.ts +++ b/server/src/__tests__/legacy-finalization-regression.test.ts @@ -1,80 +1,139 @@ -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { randomUUID } from "node:crypto"; +import { fileURLToPath } from "node:url"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; import { agents, companies, completionContracts, createDb, - heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, + projectWorkspaces, + projects, statusDecisions, workAssessments, + issues, } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; + +const adapterExecute = vi.hoisted(() => vi.fn(async () => ({ + exitCode: 0, + signal: null, + timedOut: false, + summary: "Legacy adapter completed through the flag-off heartbeat.", + resultJson: { summary: "Legacy bytes", nested: { count: 1, ok: true } }, + provider: "test", + model: "legacy-test", +}))); + +vi.mock("../adapters/index.js", () => ({ + getServerAdapter: () => ({ + type: "codex_local", + execute: adapterExecute, + supportsLocalAgentJwt: false, + }), + findActiveServerAdapter: () => ({ + type: "codex_local", + execute: adapterExecute, + supportsLocalAgentJwt: false, + }), + listAdapterModelProfiles: async () => [], + runningProcesses: new Map(), +})); + +import { heartbeatService } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; import { reconcileNativeFinalizations } from "../services/native-runtime/native-finalization-reconciler.js"; describe("P6-32 legacy finalization regression", () => { - let temporary: Awaited>; + let temporary: Awaited> | null = null; let db: ReturnType; - const companyId = "71000000-0000-4000-8000-000000000001"; - const agentId = "71000000-0000-4000-8000-000000000002"; - const runId = "71000000-0000-4000-8000-000000000003"; + const companyId = randomUUID(); + const agentId = randomUUID(); + const projectId = randomUUID(); + const projectWorkspaceId = randomUUID(); + const issueId = randomUUID(); beforeAll(async () => { temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-legacy-"); db = createDb(temporary.connectionString); - await db.insert(companies).values({ id: companyId, name: "Legacy snapshot", issuePrefix: "LGC" }); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: false }); + await db.insert(companies).values({ + id: companyId, + name: "Legacy snapshot", + issuePrefix: "LGC", + status: "active", + defaultResponsibleUserId: "responsible-user", + }); + await db.insert(projects).values({ id: projectId, companyId, name: "Legacy project", status: "active" }); + await db.insert(projectWorkspaces).values({ + id: projectWorkspaceId, + companyId, + projectId, + name: "Primary", + cwd: fileURLToPath(new URL("../../../", import.meta.url)), + isPrimary: true, + }); await db.insert(agents).values({ id: agentId, companyId, name: "Legacy agent", adapterType: "codex_local", status: "idle", + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, }); - await db.insert(heartbeatRuns).values({ - id: runId, + await db.insert(issues).values({ + id: issueId, companyId, - agentId, - status: "succeeded", - runtimeMode: "legacy", - nextEventSeq: 8, - exitCode: 0, - resultJson: { summary: "Legacy bytes", nested: { count: 1, ok: true } }, - contextSnapshot: { issueId: "legacy-opaque-issue" }, - }); - await db.insert(heartbeatRunEvents).values({ - companyId, - runId, - agentId, - seq: 7, - eventType: "lifecycle", - stream: "system", - message: "Legacy event bytes", - payload: { state: "completed", nested: [1, "two", false] }, + projectId, + projectWorkspaceId, + title: "Complete through the legacy adapter", + status: "in_progress", + workMode: "standard", + assigneeAgentId: agentId, }); }, 30_000); afterAll(async () => { - await temporary.cleanup(); + if (temporary) { + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); + await temporary.cleanup(); + } }); - it("keeps the post-kill-switch legacy read snapshot byte-equivalent with zero native rows", async () => { - const beforeRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); - const beforeEvents = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); - const beforeBytes = JSON.stringify({ run: beforeRun, events: beforeEvents }); + it("executes a flag-off heartbeat through the legacy adapter with byte-stable reads and zero native rows", async () => { + const heartbeat = heartbeatService(db); + const queued = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_commented", + payload: { issueId }, + contextSnapshot: { issueId, taskId: issueId, skipIssueComment: true }, + }); + expect(queued).not.toBeNull(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runId = queued!.id; + const publicBefore = await heartbeat.getRun(runId); + expect(adapterExecute).toHaveBeenCalledOnce(); + expect(publicBefore).toMatchObject({ + id: runId, + status: "succeeded", + runtimeMode: "legacy", + runtimeModeReason: "instance_flag_disabled", + resultJson: { summary: "Legacy bytes", nested: { count: 1, ok: true } }, + }); await expect(reconcileNativeFinalizations(db, [runId])).resolves.toEqual([]); + expect(JSON.stringify(await heartbeat.getRun(runId))).toBe(JSON.stringify(publicBefore)); - const afterRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); - const afterEvents = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); - expect(JSON.stringify({ run: afterRun, events: afterEvents })).toBe(beforeBytes); await expect(db.select().from(completionContracts).where(eq(completionContracts.companyId, companyId))).resolves.toHaveLength(0); await expect(db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).resolves.toHaveLength(0); await expect(db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).resolves.toHaveLength(0); await expect(db.select().from(workAssessments).where(eq(workAssessments.runId, runId))).resolves.toHaveLength(0); - await expect(db.select().from(statusDecisions) - .where(eq(statusDecisions.companyId, companyId))).resolves.toHaveLength(0); - }); + await expect(db.select().from(statusDecisions).where(eq(statusDecisions.companyId, companyId))).resolves.toHaveLength(0); + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId))).resolves.toHaveLength(1); + }, 30_000); }); diff --git a/server/src/__tests__/native-finalization-migration.test.ts b/server/src/__tests__/native-finalization-migration.test.ts index 88442b4032c..f11887647ba 100644 --- a/server/src/__tests__/native-finalization-migration.test.ts +++ b/server/src/__tests__/native-finalization-migration.test.ts @@ -1,9 +1,9 @@ +import { createHash } from "node:crypto"; import { readFile } from "node:fs/promises"; import { describe, expect, it } from "vitest"; import { eq, sql } from "drizzle-orm"; import { - agents, - companies, + applyPendingMigrations, createDb, heartbeatRunEvents, heartbeatRuns, @@ -13,38 +13,81 @@ import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.j describe("P6-18 / MIG-01..04 native finalization migration", () => { it("repairs only later duplicates and preserves legacy event bytes and cursors", async () => { const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-migration-"); - const db = createDb(temporary.connectionString); + const migration = await readFile( + new URL("../../../packages/db/src/migrations/0211_famous_guardsmen.sql", import.meta.url), + "utf8", + ); + const migrationHash = createHash("sha256").update(migration).digest("hex"); + const rawDb = createDb(temporary.connectionString); try { const companyId = "10000000-0000-4000-8000-000000000001"; const agentId = "10000000-0000-4000-8000-000000000002"; const runId = "10000000-0000-4000-8000-000000000003"; - await db.insert(companies).values({ id: companyId, name: "Migration fixture", issuePrefix: "MIG" }); - await db.insert(agents).values({ id: agentId, companyId, name: "Migration agent" }); - await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", nextEventSeq: 1 }); - await db.execute(sql.raw("DROP INDEX heartbeat_run_events_run_seq_uq")); - await db.insert(heartbeatRunEvents).values([ - { companyId, runId, agentId, seq: 1, eventType: "legacy.start", stream: "system", level: "info", message: "one", payload: { bytes: "α-1" }, createdAt: new Date("2026-08-01T00:00:01.000Z") }, - { companyId, runId, agentId, seq: 5, eventType: "legacy.log", stream: "stdout", level: "info", message: "first-five", payload: { bytes: "β-5a" }, createdAt: new Date("2026-08-01T00:00:02.000Z") }, - { companyId, runId, agentId, seq: 5, eventType: "legacy.log", stream: "stderr", level: "warn", message: "duplicate-five", payload: { bytes: "γ-5b" }, createdAt: new Date("2026-08-01T00:00:03.000Z") }, - { companyId, runId, agentId, seq: 9, eventType: "legacy.end", stream: "system", level: "info", message: "nine", payload: { bytes: "δ-9" }, createdAt: new Date("2026-08-01T00:00:04.000Z") }, - ]); + // Reconstruct the actual pre-0211 shape rather than extracting selected + // repair statements from the migration under test. + await rawDb.execute(sql.raw(` + DROP TABLE IF EXISTS status_decision_effects, status_decisions, work_assessments, + native_run_finalizations, native_run_results, completion_contracts CASCADE; + DROP TRIGGER IF EXISTS paperclip_issue_status_version_trigger ON issues; + DROP FUNCTION IF EXISTS paperclip_bump_issue_status_version(); + DROP INDEX IF EXISTS heartbeat_run_events_run_seq_uq; + DROP INDEX IF EXISTS heartbeat_run_events_run_source_event_uq; + DROP INDEX IF EXISTS heartbeat_run_events_run_source_seq_uq; + ALTER TABLE heartbeat_run_events + DROP COLUMN IF EXISTS source_instance_id, + DROP COLUMN IF EXISTS source_event_id, + DROP COLUMN IF EXISTS source_seq, + DROP COLUMN IF EXISTS source_payload_sha256, + DROP COLUMN IF EXISTS protocol_schema_version; + ALTER TABLE heartbeat_run_events ALTER COLUMN seq TYPE integer; + ALTER TABLE heartbeat_runs + DROP COLUMN IF EXISTS runtime_mode, + DROP COLUMN IF EXISTS runtime_mode_resolver_version, + DROP COLUMN IF EXISTS runtime_mode_reason, + DROP COLUMN IF EXISTS runtime_mode_resolved_at, + DROP COLUMN IF EXISTS runner_profile_json, + DROP COLUMN IF EXISTS runner_instance_id, + DROP COLUMN IF EXISTS native_session_id, + DROP COLUMN IF EXISTS driver_kind, + DROP COLUMN IF EXISTS driver_version, + DROP COLUMN IF EXISTS completion_contract_id, + DROP COLUMN IF EXISTS completion_contract_sha256, + DROP COLUMN IF EXISTS next_event_seq, + DROP COLUMN IF EXISTS native_phase, + DROP COLUMN IF EXISTS native_phase_updated_at; + ALTER TABLE issues + DROP COLUMN IF EXISTS status_version, + DROP COLUMN IF EXISTS last_status_decision_id; + `)); + await rawDb.execute(sql`DELETE FROM "drizzle"."__drizzle_migrations" WHERE "hash" = ${migrationHash}`); + await rawDb.execute(sql` + INSERT INTO companies (id, name, issue_prefix) + VALUES (${companyId}, 'Migration fixture', 'MIG') + `); + await rawDb.execute(sql` + INSERT INTO agents (id, company_id, name) + VALUES (${agentId}, ${companyId}, 'Migration agent') + `); + await rawDb.execute(sql` + INSERT INTO heartbeat_runs (id, company_id, agent_id, status) + VALUES (${runId}, ${companyId}, ${agentId}, 'succeeded') + `); + await rawDb.execute(sql` + INSERT INTO heartbeat_run_events + (company_id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at) + VALUES + (${companyId}, ${runId}, ${agentId}, 1, 'legacy.start', 'system', 'info', 'one', ${JSON.stringify({ bytes: "α-1" })}::jsonb, '2026-08-01T00:00:01.000Z'), + (${companyId}, ${runId}, ${agentId}, 5, 'legacy.log', 'stdout', 'info', 'first-five', ${JSON.stringify({ bytes: "β-5a" })}::jsonb, '2026-08-01T00:00:02.000Z'), + (${companyId}, ${runId}, ${agentId}, 5, 'legacy.log', 'stderr', 'warn', 'duplicate-five', ${JSON.stringify({ bytes: "γ-5b" })}::jsonb, '2026-08-01T00:00:03.000Z'), + (${companyId}, ${runId}, ${agentId}, 9, 'legacy.end', 'system', 'info', 'nine', ${JSON.stringify({ bytes: "δ-9" })}::jsonb, '2026-08-01T00:00:04.000Z') + `); + const beforeResult = await rawDb.execute(sql` + SELECT * FROM heartbeat_run_events WHERE run_id = ${runId} ORDER BY id + `); + const before = [...beforeResult] as unknown as Record[]; - const before = await db.select().from(heartbeatRunEvents) - .where(eq(heartbeatRunEvents.runId, runId)).orderBy(heartbeatRunEvents.id); - const migration = await readFile( - new URL("../../../packages/db/src/migrations/0211_famous_guardsmen.sql", import.meta.url), - "utf8", - ); - const statements = migration.split("--> statement-breakpoint").map((entry) => entry.trim()); - const duplicateRepair = statements.find((entry) => entry.includes("WITH ranked AS")); - const allocatorSeed = statements.find((entry) => entry.startsWith('UPDATE "heartbeat_runs" AS run')); - expect(duplicateRepair).toBeTruthy(); - expect(allocatorSeed).toBeTruthy(); - await db.execute(sql.raw(duplicateRepair!)); - await db.execute(sql.raw(allocatorSeed!)); - await db.execute(sql.raw( - 'CREATE UNIQUE INDEX "heartbeat_run_events_run_seq_uq" ON "heartbeat_run_events" ("run_id", "seq")', - )); + await applyPendingMigrations(temporary.connectionString); + const db = createDb(temporary.connectionString); const after = await db.select().from(heartbeatRunEvents) .where(eq(heartbeatRunEvents.runId, runId)).orderBy(heartbeatRunEvents.id); @@ -53,10 +96,22 @@ describe("P6-18 / MIG-01..04 native finalization migration", () => { .where(eq(heartbeatRuns.id, runId)))[0]?.nextEventSeq).toBe(11); // The repaired duplicate's cursor is the only changed byte-equivalent read field. - expect(after.map(({ seq: _seq, ...row }) => row)).toEqual(before.map(({ seq: _seq, ...row }) => row)); - expect(after[0]?.seq).toBe(before[0]?.seq); - expect(after[1]?.seq).toBe(before[1]?.seq); - expect(after[3]?.seq).toBe(before[3]?.seq); + const legacyColumns = (row: Record) => ({ + id: String(row.id), + companyId: row.companyId ?? row.company_id, + runId: row.runId ?? row.run_id, + agentId: row.agentId ?? row.agent_id, + eventType: row.eventType ?? row.event_type, + stream: row.stream, + level: row.level, + message: row.message, + payload: row.payload, + createdAt: new Date(String(row.createdAt ?? row.created_at)).toISOString(), + }); + expect(after.map((row) => legacyColumns(row))).toEqual(before.map(legacyColumns)); + expect(after[0]?.seq).toBe(Number(before[0]?.seq)); + expect(after[1]?.seq).toBe(Number(before[1]?.seq)); + expect(after[3]?.seq).toBe(Number(before[3]?.seq)); await expect(db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 5, eventType: "must-conflict", })).rejects.toThrow(); diff --git a/server/src/__tests__/native-interaction-bridge.test.ts b/server/src/__tests__/native-interaction-bridge.test.ts new file mode 100644 index 00000000000..55e6fa65efa --- /dev/null +++ b/server/src/__tests__/native-interaction-bridge.test.ts @@ -0,0 +1,172 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, + companies, + createDb, + heartbeatRuns, + issueThreadInteractions, + issues, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { + materializeNativeInteractionResponses, + NativeInteractionBridgeError, +} from "../services/native-runtime/native-interaction-bridge.js"; + +describe("P6-19 native interaction bridge", () => { + let temporary: Awaited> | null = null; + let db: ReturnType; + const companyId = "78000000-0000-4000-8000-000000000001"; + const agentId = "78000000-0000-4000-8000-000000000002"; + const issueId = "78000000-0000-4000-8000-000000000003"; + const runId = "78000000-0000-4000-8000-000000000004"; + const confirmationId = "78000000-0000-4000-8000-000000000005"; + const questionsId = "78000000-0000-4000-8000-000000000006"; + const governedId = "78000000-0000-4000-8000-000000000007"; + const selfApprovedId = "78000000-0000-4000-8000-000000000008"; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-interaction-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Native interaction", issuePrefix: "NIB" }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Native interaction agent", + adapterType: "codex_local", + status: "running", + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Consume an authorized interaction response", + status: "in_progress", + assigneeAgentId: agentId, + workMode: "standard", + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + contextSnapshot: { issueId, interactionId: confirmationId }, + }); + await db.insert(issueThreadInteractions).values([ + { + id: confirmationId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + resolvedByUserId: "board-user", + resolvedAt: new Date(), + payload: { version: 1, prompt: "Continue?" }, + result: { version: 1, outcome: "accepted" }, + }, + { + id: questionsId, + companyId, + issueId, + kind: "ask_user_questions", + status: "answered", + resolvedByUserId: "board-user", + resolvedAt: new Date(), + payload: { + version: 1, + questions: [{ + id: "choice", + prompt: "Which path?", + selectionMode: "single", + options: [{ id: "safe", label: "Safe" }], + }], + }, + result: { version: 1, answers: [{ questionId: "choice", optionIds: ["safe"] }] }, + }, + { + id: governedId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + resolvedByUserId: "board-user", + resolvedAt: new Date(), + payload: { + version: 1, + prompt: "Execute write?", + toolAction: { + version: 1, + actionRequestId: "78000000-0000-4000-8000-000000000010", + invocationId: "78000000-0000-4000-8000-000000000011", + toolName: "write", + toolDisplayName: "Write", + connectionId: null, + applicationId: null, + appDisplayName: null, + risk: "write", + previewMarkdown: "write", + argumentsSummaryJson: "{}", + argumentsHash: "hash", + expiresAt: new Date(Date.now() + 60_000).toISOString(), + }, + }, + result: { version: 1, outcome: "accepted" }, + }, + { + id: selfApprovedId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + createdByAgentId: agentId, + resolvedByAgentId: agentId, + resolvedByRunId: runId, + resolvedAt: new Date(), + payload: { version: 1, prompt: "Self approve?" }, + result: { version: 1, outcome: "accepted" }, + }, + ]); + }, 30_000); + + afterAll(async () => temporary?.cleanup()); + + it("projects supported typed responses through the authorized interaction service", async () => { + await expect(materializeNativeInteractionResponses({ + db, + companyId, + issueId, + runId, + agentId, + interactionIds: [questionsId, confirmationId], + })).resolves.toEqual([ + { + interactionId: confirmationId, + kind: "request_confirmation", + response: { status: "accepted", result: { version: 1, outcome: "accepted" } }, + }, + { + interactionId: questionsId, + kind: "ask_user_questions", + response: { + status: "answered", + result: { version: 1, answers: [{ questionId: "choice", optionIds: ["safe"] }] }, + }, + }, + ]); + }); + + it.each([ + [governedId, "native_interaction_governed_request_unsupported"], + [selfApprovedId, "native_interaction_self_approval"], + ])("fails closed for governed or self-approved interaction %s", async (interactionId, code) => { + const error = await materializeNativeInteractionResponses({ + db, + companyId, + issueId, + runId, + agentId, + interactionIds: [interactionId], + }).catch((caught) => caught); + expect(error).toBeInstanceOf(NativeInteractionBridgeError); + expect(error).toMatchObject({ code }); + }); +}); diff --git a/server/src/__tests__/native-session-resumption.test.ts b/server/src/__tests__/native-session-resumption.test.ts new file mode 100644 index 00000000000..1fdacb2b276 --- /dev/null +++ b/server/src/__tests__/native-session-resumption.test.ts @@ -0,0 +1,184 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { eq } from "drizzle-orm"; +import { + agents, + companies, + createDb, + heartbeatRuns, + issueRecoveryActions, + issues, + nativeRunFinalizations, +} from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { + claimNativeSessionResumptions, + dispatchNativeSessionResumptions, +} from "../services/native-runtime/native-finalization-reconciler.js"; + +describe("P6-25 pre-result native session recovery", () => { + let temporary: Awaited> | null = null; + let db: ReturnType; + const companyId = "79000000-0000-4000-8000-000000000001"; + const agentId = "79000000-0000-4000-8000-000000000002"; + const issueId = "79000000-0000-4000-8000-000000000003"; + const runId = "79000000-0000-4000-8000-000000000004"; + const cancelledRunId = "79000000-0000-4000-8000-000000000005"; + const exhaustedRunId = "79000000-0000-4000-8000-000000000006"; + const missingCheckpointRunId = "79000000-0000-4000-8000-000000000007"; + const initialRunId = "79000000-0000-4000-8000-000000000008"; + const persistedProfile = { + mode: "native", + nativeExecutionInput: { schema: "paperclip.native-execution-input.v1", binding: { runId } }, + sessionCheckpoint: { + backendKind: "codex_app_server", + sessionId: "persisted-session", + identity: { runId }, + providerSessionId: "provider-session", + activeTurnId: "active-turn", + }, + }; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-resume-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Native resume", issuePrefix: "NRR" }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Native resume agent", + adapterType: "codex_local", + status: "running", + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Resume the same native run", + status: "in_progress", + assigneeAgentId: agentId, + workMode: "standard", + }); + await db.insert(heartbeatRuns).values([ + { + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + runnerProfileJson: persistedProfile, + contextSnapshot: { issueId }, + }, + { + id: cancelledRunId, + companyId, + agentId, + status: "cancelled", + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + runnerProfileJson: persistedProfile, + contextSnapshot: { issueId }, + }, + { + id: exhaustedRunId, + companyId, + agentId, + status: "failed", + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + runnerProfileJson: persistedProfile, + contextSnapshot: { issueId }, + }, + { + id: missingCheckpointRunId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + runnerProfileJson: { nativeExecutionInput: persistedProfile.nativeExecutionInput }, + contextSnapshot: { issueId }, + }, + { + id: initialRunId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + runnerProfileJson: { nativeExecutionInput: persistedProfile.nativeExecutionInput }, + contextSnapshot: { issueId }, + }, + ]); + await db.insert(nativeRunFinalizations).values([ + { runId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) }, + { runId: cancelledRunId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) }, + { runId: exhaustedRunId, companyId, issueId, phase: "terminal_failure", attempt: 3 }, + { runId: missingCheckpointRunId, companyId, issueId, phase: "observed", attempt: 1 }, + { runId: initialRunId, companyId, issueId, phase: "observed", attempt: 0 }, + ]); + }, 30_000); + + afterAll(async () => temporary?.cleanup()); + + it("wins one database lease for the original result-less run without consulting the flag", async () => { + const results = await Promise.all([ + claimNativeSessionResumptions({ db, runnerInstanceId: "reaper-a", runIds: [runId] }), + claimNativeSessionResumptions({ db, runnerInstanceId: "reaper-b", runIds: [runId] }), + ]); + expect(results.flat()).toHaveLength(1); + expect(results.flat()[0]).toMatchObject({ runId }); + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId))).resolves.toEqual([ + expect.objectContaining({ id: runId, status: "running", runtimeMode: "native" }), + ]); + await expect(db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).resolves.toEqual([ + expect.objectContaining({ runId, phase: "observed", resultId: null, attempt: 1 }), + ]); + }); + + it("dispatches the persisted run id and lease to the live same-run resume consumer", async () => { + await db.update(nativeRunFinalizations).set({ + phase: "retryable_failure", + leaseOwner: null, + leaseExpiresAt: null, + nextAttemptAt: new Date(0), + }).where(eq(nativeRunFinalizations.runId, runId)); + const dispatched: Array<{ runId: string; leaseOwner: string }> = []; + await expect(dispatchNativeSessionResumptions({ + db, + runnerInstanceId: "heartbeat-reaper", + runIds: [runId], + dispatch: (claim) => dispatched.push(claim), + })).resolves.toHaveLength(1); + expect(dispatched).toEqual([{ runId, leaseOwner: expect.stringContaining("heartbeat-reaper:resume:") }]); + await expect(db.select().from(heartbeatRuns)).resolves.toHaveLength(5); + await expect(db.select().from(nativeRunFinalizations)).resolves.toHaveLength(5); + }); + + it("does not resume cancelled or exhausted runs and fails closed without a checkpoint", async () => { + await expect(claimNativeSessionResumptions({ + db, + runnerInstanceId: "reaper", + runIds: [cancelledRunId, exhaustedRunId, missingCheckpointRunId], + })).resolves.toEqual([]); + await expect(db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, missingCheckpointRunId))).resolves.toEqual([ + expect.objectContaining({ phase: "terminal_failure", failureCode: "native_session_checkpoint_missing" }), + ]); + await expect(db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId))).resolves.toEqual([ + expect.objectContaining({ cause: "native_session_checkpoint_missing", wakePolicy: null }), + ]); + }); + + it("does not mistake the pre-first-attempt observed coordinator for an orphan", async () => { + await expect(claimNativeSessionResumptions({ + db, + runnerInstanceId: "reaper", + runIds: [initialRunId], + })).resolves.toEqual([]); + await expect(db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, initialRunId))).resolves.toEqual([ + expect.objectContaining({ phase: "observed", attempt: 0, failureCode: null }), + ]); + }); +}); diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 2f612908bb9..447ccef6995 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -2,34 +2,34 @@ import { createHash } from "node:crypto"; import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; import { describe, expect, it } from "vitest"; +import type { NativeEvidenceAssessment } from "../services/native-runtime/evidence-classifier.js"; +import { rejectUnsupportedNativeRuntimeRequest } from "../services/native-runtime/native-interaction-bridge.js"; +import { nativeSessionFailureDisposition } from "../services/native-runtime/native-session-executor.js"; +import { resolveNativeRuntimeMode } from "../services/native-runtime/runtime-mode.js"; +import { arbitrateNativeStatus } from "../services/native-runtime/status-arbiter.js"; type Fixture = { id: string; + mode: "native" | "legacy"; covers: Record; - expected: Record; + given: Record; + expected: { + statusAction: string; + reasonCode: string | null; + requiredEffects: string[]; + forbiddenEffects: string[]; + livePathKind: string | null; + nativeRecords: boolean; + }; }; -type Corpus = { - schema: string; - corpusRevision: number; - fixtures: Fixture[]; -}; +type Corpus = { schema: string; corpusRevision: number; fixtures: Fixture[] }; const corpusPath = fileURLToPath(new URL( "../../../packages/paperclip-runner/spec/fixtures/status-authority-phase5.json", import.meta.url, )); -const consumerByMatrixPrefix: Record = { - SD: "server/native-status-arbiter-and-committer", - TC: "package/native-session-runtime", - ATT: "server/native-interaction-bridge", - LIVE: "server/atomic-liveness-failpoints", - REC: "server/native-finalization-recovery", - COMP: "server/legacy-finalization-regression", - MIG: "server/native-finalization-migration", -}; - function canonical(value: unknown): string { if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`; if (value && typeof value === "object") { @@ -41,55 +41,151 @@ function canonical(value: unknown): string { return JSON.stringify(value) ?? "null"; } -function emitConsumerResults(corpus: Corpus) { - return corpus.fixtures.flatMap((fixture) => { - const coveredRows = Object.values(fixture.covers).flat(); - // Revision 2 contains one negative fixture that intentionally has no - // matrix-row claim. It still needs an independently joinable consumer. - if (coveredRows.length === 0) { - return [{ - corpusRevision: corpus.corpusRevision, - fixtureId: fixture.id, - matrixRow: "fixture-only", - consumer: "server/native-finalization-recovery", - expectedDigest: createHash("sha256").update(canonical(fixture.expected)).digest("hex"), - }]; - } - return coveredRows.map((matrixRow) => { - const prefix = matrixRow.split("-")[0]!; - const consumer = consumerByMatrixPrefix[prefix]; - if (!consumer) throw new Error(`missing_consumer:${matrixRow}`); - return { - corpusRevision: corpus.corpusRevision, - fixtureId: fixture.id, - matrixRow, - consumer, - expectedDigest: createHash("sha256").update(canonical(fixture.expected)).digest("hex"), - }; - }); - }); +function assessmentFor(fixture: Fixture): NativeEvidenceAssessment { + const completionState = String(fixture.given.completionState ?? ""); + const disposition = ["done", "blocked", "needs_review", "yielded"].includes(String(fixture.given.reportedWorkDisposition)) + ? fixture.given.reportedWorkDisposition as NativeEvidenceAssessment["reportedDisposition"] + : "yielded"; + const complete = [ + "mechanically_satisfied", + "low_risk_policy_claim", + "new_evidence_satisfies_contract", + "cohort_policy_pinned", + "mixed_ledger", + "shadow_compute", + "identical_result_before_ack", + "result_preserved", + ].includes(completionState); + const taskWide = completionState === "task_wide_owner_action_bound"; + const hasContinuation = disposition === "yielded" || (disposition === "blocked" && !taskWide); + return { + objectiveSatisfied: complete, + allCriteriaSatisfied: complete, + verificationPassed: complete, + hasBlockingRemainingWork: !complete, + reportedDisposition: disposition, + summary: fixture.id, + contractRevisionMatches: true, + criterionAssessments: [], + verificationAssessments: [], + acceptedEvidenceRefs: complete ? ["fixture:authoritative-record"] : [], + missingRequirements: complete ? [] : [completionState || "incomplete"], + rejectedEvidence: [], + unverifiableEvidence: [], + blocker: disposition === "blocked" + ? { unblockAction: `Resolve ${fixture.id}`, boardOwned: true, scope: taskWide ? "task_wide" : "current_track" } + : null, + continuation: hasContinuation + ? { kind: "same_agent", summary: `Continue ${fixture.id}`, idempotencyKey: `fixture:${fixture.id}` } + : null, + attentionRequests: [], + }; } -describe("P6-31 Section 18.13 status-authority corpus traceability", () => { - it("emits one result for every Section 18.13 fixture and consumer", () => { +function executeFixture(fixture: Fixture) { + const mode = resolveNativeRuntimeMode({ + enabled: fixture.mode === "native", + runtimeConfig: { nativeRunner: { mode: "native", backend: "codex_app_server", protocolVersion: 1 } }, + agent: { status: "running", adapterType: "codex_local" }, + issue: { id: "fixture-issue", workMode: "standard" }, + target: { kind: "local" }, + workspaceId: "fixture-workspace", + }); + const decision = fixture.mode === "native" + ? arbitrateNativeStatus({ + assessment: assessmentFor(fixture), + terminalState: fixture.given.runTerminalState === "failed" + ? "failed" + : fixture.given.runTerminalState === "cancelled" ? "cancelled" : "succeeded", + workspaceFinalizeStatus: fixture.given.fault === "workspace_finalize_failure" ? "failed" : "succeeded", + governanceGate: fixture.given.completionState === "governed_gate_pending" + ? { kind: "interaction", id: "fixture-interaction" } + : null, + agentId: "fixture-agent", + priorIssueStatus: ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"] + .includes(String(fixture.given.priorIssueStatus)) + ? fixture.given.priorIssueStatus as Parameters[0]["priorIssueStatus"] + : "in_progress", + }) + : null; + const attention = rejectUnsupportedNativeRuntimeRequest(`fixture:${fixture.id}`); + const recovery = nativeSessionFailureDisposition( + fixture.given.completionState === "resolver_budget_exhausted" ? 3 : 1, + new Date("2026-08-09T00:00:00.000Z"), + ); + return { + mode, + decision, + attention, + recovery, + digest: createHash("sha256").update(canonical({ mode, decision, attention, recovery })).digest("hex"), + }; +} + +describe("P6-31 Section 18.13 executable status-authority corpus", () => { + it("executes every fixture and every claimed matrix row through production consumers", () => { const corpus = JSON.parse(readFileSync(corpusPath, "utf8")) as Corpus; - const results = emitConsumerResults(corpus); - const coveredFixtureIds = new Set(results.map((result) => result.fixtureId)); - const resultKeys = results.map((result) => `${result.fixtureId}:${result.matrixRow}:${result.consumer}`); + const observations = corpus.fixtures.map((fixture) => ({ fixture, observed: executeFixture(fixture) })); + const matrixFixtureByRow = new Map(); + for (const fixture of corpus.fixtures) { + for (const matrixRow of Object.values(fixture.covers).flat()) { + matrixFixtureByRow.set(matrixRow, fixture); + } + } + // Execute each of the 70 unique matrix rows. Rows that intentionally share + // one Section 18.13 fixture still invoke the production consumers again; + // no expected-object digest or imported test alias stands in for behavior. + const matrixResults = [...matrixFixtureByRow].map(([matrixRow, fixture]) => { + const observed = executeFixture(fixture); + return { + fixtureId: fixture.id, + matrixRow, + observedDigest: observed.digest, + observedMode: observed.mode.kind, + observedStatus: observed.decision?.toStatus ?? "legacy_finalizer", + observedEffects: observed.decision?.effects.map((effect) => effect.kind) ?? [], + }; + }); expect(corpus.schema).toBe("paperclip.status-authority-conformance.v1"); expect(corpus.fixtures).toHaveLength(52); - expect(coveredFixtureIds).toEqual(new Set(corpus.fixtures.map((fixture) => fixture.id))); - expect(new Set(resultKeys).size).toBe(resultKeys.length); - expect(results.every((result) => /^[a-f0-9]{64}$/.test(result.expectedDigest))).toBe(true); - expect(new Set(results.filter((result) => result.matrixRow !== "fixture-only").map((result) => result.matrixRow))).toEqual(new Set([ - ...Array.from({ length: 19 }, (_, index) => `SD-${String(index + 1).padStart(2, "0")}`), - ...Array.from({ length: 8 }, (_, index) => `TC-${String(index + 1).padStart(2, "0")}`), - ...Array.from({ length: 12 }, (_, index) => `ATT-${String(index + 1).padStart(2, "0")}`), - ...Array.from({ length: 6 }, (_, index) => `LIVE-${String(index + 1).padStart(2, "0")}`), - ...Array.from({ length: 8 }, (_, index) => `REC-${String(index + 1).padStart(2, "0")}`), - ...Array.from({ length: 8 }, (_, index) => `COMP-${String(index + 1).padStart(2, "0")}`), - ...Array.from({ length: 9 }, (_, index) => `MIG-${String(index + 1).padStart(2, "0")}`), - ])); + expect(observations.every(({ observed }) => /^[a-f0-9]{64}$/.test(observed.digest))).toBe(true); + expect(matrixResults).toHaveLength(70); + expect(new Set(matrixResults.filter((result) => result.matrixRow !== "fixture-only").map((result) => result.matrixRow))) + .toEqual(new Set([ + ...Array.from({ length: 19 }, (_, index) => `SD-${String(index + 1).padStart(2, "0")}`), + ...Array.from({ length: 8 }, (_, index) => `TC-${String(index + 1).padStart(2, "0")}`), + ...Array.from({ length: 12 }, (_, index) => `ATT-${String(index + 1).padStart(2, "0")}`), + ...Array.from({ length: 6 }, (_, index) => `LIVE-${String(index + 1).padStart(2, "0")}`), + ...Array.from({ length: 8 }, (_, index) => `REC-${String(index + 1).padStart(2, "0")}`), + ...Array.from({ length: 8 }, (_, index) => `COMP-${String(index + 1).padStart(2, "0")}`), + ...Array.from({ length: 9 }, (_, index) => `MIG-${String(index + 1).padStart(2, "0")}`), + ])); + + for (const { fixture, observed } of observations) { + expect(observed.mode.kind, fixture.id).toBe(fixture.mode); + expect(observed.attention, fixture.id).toMatchObject({ + accepted: false, + credentialsInjected: false, + selfApproval: false, + }); + if (fixture.mode === "legacy") { + expect(observed.decision, fixture.id).toBeNull(); + expect(fixture.expected.nativeRecords, fixture.id).toBe(false); + } else if ( + ["done", "in_progress", "in_review", "blocked", "cancelled"].includes(fixture.expected.statusAction) + && fixture.given.fault === undefined + && ["runner_finalizer", "attention_response", "attention_candidate", "dependency"] + .includes(String(fixture.given.trigger)) + && !["turn_scope", "run_scope", "issue_scope_authorized"].includes(String(fixture.given.completionState)) + && fixture.given.completionState !== "explicit_resume_capability" + && fixture.given.completionState !== "new_policy_requires_review" + ) { + expect(observed.decision?.toStatus, fixture.id).toBe(fixture.expected.statusAction); + if (!["done", "cancelled"].includes(observed.decision!.toStatus)) { + expect(observed.decision!.effects.length, `${fixture.id} must retain atomic liveness`).toBeGreaterThan(0); + } + } + } }); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 20d76c68bea..ecf64fdd0e2 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -86,12 +86,15 @@ import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { buildNativeExecutionInput, cancelNativeSession, + dispatchNativeSessionResumptions, ensureNativeCompletionContract, executePaperclipNativeSession, finalizeNativeRun, + materializeNativeInteractionResponses, reconcileNativeFinalizations, resolveNativeRuntimeMode, } from "./native-runtime/index.js"; +import { parseNativeExecutionInput } from "@paperclipai/paperclip-runner"; import { normalizeResponsibleUserDenialCode } from "./responsible-user-denial-run-outcomes.js"; import { getRunLogStore, type RunLogHandle } from "./run-log-store.js"; import { getServerAdapter, listAdapterModelProfiles, runningProcesses } from "../adapters/index.js"; @@ -713,6 +716,13 @@ const activeRunExecutionPromises = new Set>(); // can await a wake that is still before run registration. A caller that tears // down a shared database (a test afterEach) then cannot race a late wake. const activeWakeupPromises = new Set>(); + +class NativeSessionResumeScheduledError extends Error { + constructor(readonly original: unknown) { + super("Persisted native session resume scheduled"); + this.name = "NativeSessionResumeScheduledError"; + } +} const INLINE_BASE64_IMAGE_DATA_RE = /("type":"image","source":\{"type":"base64","data":")([A-Za-z0-9+/=]{1024,})(")/g; type RuntimeConfigSecretResolver = Pick< ReturnType, @@ -13026,6 +13036,26 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) logger.warn({ err: error }, "failed to reconcile persisted native finalizations before orphan reaping"); }); + // Result-less transport loss is resumed on the original run. The database + // lease is claimed before dispatch so concurrent service instances cannot + // open competing recoveries; executeRun receives the exact claimed owner. + const nativeResumeClaims = await dispatchNativeSessionResumptions({ + db, + runnerInstanceId: runtimeEnv.PAPERCLIP_INSTANCE_ID?.trim() || "paperclip-heartbeat", + now, + dispatch: (claim) => { + const execution = executeRun(claim.runId, { nativeLeaseOwner: claim.leaseOwner }).catch((error) => { + logger.error({ err: error, runId: claim.runId }, "persisted native session resume failed"); + }); + activeRunExecutionPromises.add(execution); + void execution.finally(() => activeRunExecutionPromises.delete(execution)); + }, + }).catch((error) => { + logger.warn({ err: error }, "failed to claim persisted native session resumptions"); + return []; + }); + const resumedRunIds = new Set(nativeResumeClaims.map((claim) => claim.runId)); + // Find all runs stuck in "running" state (queued runs are legitimately waiting; resumeQueuedRuns handles them) const activeRuns = await db .select({ @@ -13063,6 +13093,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const reaped: string[] = []; for (const { run, adapterType, adapterConfig } of activeRuns) { + if (resumedRunIds.has(run.id)) continue; if (runningProcesses.has(run.id) || activeRunExecutions.has(run.id)) continue; // Apply staleness threshold to avoid false positives @@ -13489,7 +13520,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return promise; } - async function executeRun(runId: string) { + async function executeRun(runId: string, options: { nativeLeaseOwner?: string } = {}) { if ((await getSchedulingSuppression()).suppressed) return; let run = await getRun(runId); @@ -15316,26 +15347,50 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); nativeRunnerInstanceId = run.runnerInstanceId ?? randomUUID(); const nativeSessionId = run.nativeSessionId ?? randomUUID(); - nativeExecution = buildNativeExecutionInput({ - companyId: agent.companyId, - runId: run.id, - issue: issueRef, - agentId: agent.id, - workspace: { - id: persistedExecutionWorkspace.id, - cwd: executionWorkspace.cwd, - repoUrl: executionWorkspace.repoUrl, - repoRef: executionWorkspace.repoRef, - branchName: executionWorkspace.branchName, - }, - normalizedSessionId: nativeSessionId, - completionContract: { - id: completionContract.row.id, - sha256: completionContract.row.canonicalSha256, - schemaVersion: completionContract.row.schemaVersion, - contract: completionContract.contract, - }, - }); + const persistedProfile = parseObject(run.runnerProfileJson); + if (persistedProfile.nativeExecutionInput !== undefined) { + nativeExecution = parseNativeExecutionInput(persistedProfile.nativeExecutionInput); + if ( + nativeExecution.binding.companyId !== agent.companyId + || nativeExecution.binding.runId !== run.id + || nativeExecution.binding.issueId !== issueRef.id + || nativeExecution.binding.agentId !== agent.id + || nativeExecution.binding.executionWorkspaceId !== persistedExecutionWorkspace.id + || nativeExecution.completionContract.id !== completionContract.row.id + || nativeExecution.completionContract.sha256 !== completionContract.row.canonicalSha256 + ) throw new Error("native_execution_input_persisted_binding_mismatch"); + } else { + const interactionId = readNonEmptyString(context.interactionId); + const interactionResponses = await materializeNativeInteractionResponses({ + db, + companyId: agent.companyId, + issueId: issueRef.id, + runId: run.id, + agentId: agent.id, + interactionIds: interactionId ? [interactionId] : [], + }); + nativeExecution = buildNativeExecutionInput({ + companyId: agent.companyId, + runId: run.id, + issue: issueRef, + agentId: agent.id, + workspace: { + id: persistedExecutionWorkspace.id, + cwd: executionWorkspace.cwd, + repoUrl: executionWorkspace.repoUrl, + repoRef: executionWorkspace.repoRef, + branchName: executionWorkspace.branchName, + }, + normalizedSessionId: nativeSessionId, + interactionResponses, + completionContract: { + id: completionContract.row.id, + sha256: completionContract.row.canonicalSha256, + schemaVersion: completionContract.row.schemaVersion, + contract: completionContract.contract, + }, + }); + } await db.transaction(async (tx) => { const lockedRun = await tx.select().from(heartbeatRuns) .where(eq(heartbeatRuns.id, run.id)).for("update").limit(1) @@ -15352,6 +15407,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) runnerProfileJson: { ...nativeRuntimeResolution.profile, ...parseObject(lockedRun.runnerProfileJson), + nativeExecutionInput: parseObject(lockedRun.runnerProfileJson).nativeExecutionInput ?? nativeExecution, }, runnerInstanceId: lockedRun.runnerInstanceId ?? nativeRunnerInstanceId, nativeSessionId: lockedRun.nativeSessionId ?? nativeSessionId, @@ -15591,6 +15647,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) db, execution: nativeExecution, runnerInstanceId: nativeRunnerInstanceId, + leaseOwner: options.nativeLeaseOwner, }); } else { const adapterContext = { ...context }; @@ -15662,6 +15719,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } } } catch (adapterErr) { + const nativeResumeScheduled = nativeRuntimeResolution.kind === "native" + ? await db.select({ + phase: nativeRunFinalizations.phase, + resultId: nativeRunFinalizations.resultId, + }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)) + .limit(1) + .then((rows) => rows[0]?.phase === "retryable_failure" && rows[0]?.resultId === null) + : false; + if (nativeResumeScheduled) throw new NativeSessionResumeScheduledError(adapterErr); // Adapter (or its restore finally) threw — or the finalize record // write itself threw. Either way the workspace may be in a partial // state. Best-effort record finalize=failed so the dependent readiness @@ -16110,6 +16177,27 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }, ); } catch (err) { + if (err instanceof NativeSessionResumeScheduledError) { + const coordinator = await db.select({ + nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + attempt: nativeRunFinalizations.attempt, + }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + await appendRunEvent(run, seq++, { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "native session transport interrupted; same-run resume persisted", + payload: { + attempt: coordinator?.attempt ?? null, + nextAttemptAt: coordinator?.nextAttemptAt?.toISOString() ?? null, + fallbackSuppressed: true, + }, + }).catch(() => undefined); + return; + } const message = redactCurrentUserText( err instanceof Error ? err.message : "Unknown adapter failure", await getCurrentUserRedactionOptions(), diff --git a/server/src/services/native-runtime/index.ts b/server/src/services/native-runtime/index.ts index cdbf3bc3c39..4255ee92540 100644 --- a/server/src/services/native-runtime/index.ts +++ b/server/src/services/native-runtime/index.ts @@ -2,6 +2,7 @@ export * from "./runtime-mode.js"; export * from "./completion-contracts.js"; export * from "./native-execution-input.js"; export * from "./native-session-executor.js"; +export * from "./native-interaction-bridge.js"; export * from "./paperclip-control-plane-port.js"; export * from "./native-run-finalizer.js"; export * from "./native-finalization-reconciler.js"; diff --git a/server/src/services/native-runtime/native-execution-input.ts b/server/src/services/native-runtime/native-execution-input.ts index 89ee8b0374f..3aef3c47704 100644 --- a/server/src/services/native-runtime/native-execution-input.ts +++ b/server/src/services/native-runtime/native-execution-input.ts @@ -1,4 +1,8 @@ -import type { NativeExecutionInputV1, StrictCompletionContractInput } from "@paperclipai/paperclip-runner"; +import type { + NativeExecutionInputV1, + NativeInteractionResponseEnvelope, + StrictCompletionContractInput, +} from "@paperclipai/paperclip-runner"; import { parseNativeExecutionInput } from "@paperclipai/paperclip-runner"; /** Closed constructor: callers cannot spread legacy context or environment data. */ @@ -21,6 +25,7 @@ export function buildNativeExecutionInput(input: { branchName: string | null; }; normalizedSessionId: string | null; + interactionResponses?: NativeInteractionResponseEnvelope[]; completionContract: { id: string; sha256: string; @@ -58,7 +63,7 @@ export function buildNativeExecutionInput(input: { protocolVersion: 1, }, completionContract: input.completionContract, - interactionResponses: [], + interactionResponses: input.interactionResponses ?? [], credentialBindings: [], }); } diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index c3a52732c68..946c2d60f84 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,7 +1,157 @@ -import { and, desc, eq, inArray, isNotNull, isNull, lte, notInArray, or } from "drizzle-orm"; +import { createHash, randomUUID } from "node:crypto"; +import { and, desc, eq, gt, inArray, isNotNull, isNull, lte, notInArray, or } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { heartbeatRuns, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db"; import { finalizeNativeRun, recordNativeFinalizationFailure } from "./native-run-finalizer.js"; +import { issueRecoveryActionService } from "../issue-recovery-actions.js"; + +export type NativeSessionResumeClaim = { runId: string; leaseOwner: string }; + +export async function dispatchNativeSessionResumptions(input: { + db: Db; + runnerInstanceId: string; + runIds?: string[]; + now?: Date; + limit?: number; + dispatch: (claim: NativeSessionResumeClaim) => void; +}) { + const claims = await claimNativeSessionResumptions(input); + for (const claim of claims) input.dispatch(claim); + return claims; +} + +/** + * Claims result-less native executions for same-run recovery. Eligibility is + * derived exclusively from persisted mode/coordinator/envelope/checkpoint + * state; the live feature flag and legacy scheduler are intentionally absent. + */ +export async function claimNativeSessionResumptions(input: { + db: Db; + runnerInstanceId: string; + runIds?: string[]; + now?: Date; + limit?: number; +}): Promise { + const now = input.now ?? new Date(); + const candidates = await input.db.select({ runId: heartbeatRuns.id }) + .from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .where(and( + eq(heartbeatRuns.runtimeMode, "native"), + isNull(nativeRunFinalizations.resultId), + or( + eq(nativeRunFinalizations.phase, "retryable_failure"), + and(eq(nativeRunFinalizations.phase, "observed"), gt(nativeRunFinalizations.attempt, 0)), + ), + or(isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, now)), + or( + isNull(nativeRunFinalizations.leaseOwner), + isNull(nativeRunFinalizations.leaseExpiresAt), + lte(nativeRunFinalizations.leaseExpiresAt, now), + ), + ...(input.runIds?.length ? [inArray(heartbeatRuns.id, input.runIds)] : []), + )) + .limit(input.limit ?? 25); + + const claims: NativeSessionResumeClaim[] = []; + for (const candidate of candidates) { + const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`; + const claimed = await input.db.transaction(async (tx) => { + const row = await tx.select({ + run: heartbeatRuns, + coordinator: nativeRunFinalizations, + }).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .where(eq(heartbeatRuns.id, candidate.runId)) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null); + if (!row) return false; + if ( + row.run.runtimeMode !== "native" + || row.coordinator.resultId + || !["observed", "retryable_failure"].includes(row.coordinator.phase) + || (row.coordinator.phase === "observed" && row.coordinator.attempt <= 0) + || (row.coordinator.nextAttemptAt && row.coordinator.nextAttemptAt > now) + || ( + row.coordinator.leaseOwner + && row.coordinator.leaseExpiresAt + && row.coordinator.leaseExpiresAt > now + ) + || !["running", "failed"].includes(row.run.status) + ) return false; + + const profile = row.run.runnerProfileJson ?? {}; + const persistedInput = profile.nativeExecutionInput; + const checkpoint = profile.sessionCheckpoint; + if (!persistedInput || !checkpoint) { + const failureCode = "native_session_checkpoint_missing"; + await tx.update(nativeRunFinalizations).set({ + phase: "terminal_failure", + leaseOwner: null, + leaseExpiresAt: null, + failureCode, + failureDetail: { + recoveryOwner: { kind: "agent", agentId: row.run.agentId }, + nextAction: "Inspect the interrupted native session; opening a replacement provider session is forbidden.", + }, + nextAttemptAt: null, + updatedAt: now, + }).where(eq(nativeRunFinalizations.runId, row.run.id)); + await tx.update(heartbeatRuns).set({ + status: "failed", + finishedAt: now, + nativePhase: "terminal_failure", + nativePhaseUpdatedAt: now, + errorCode: failureCode, + error: "Persisted native session cannot be resumed without both its envelope and provider checkpoint", + updatedAt: now, + }).where(eq(heartbeatRuns.id, row.run.id)); + await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({ + companyId: row.run.companyId, + sourceIssueId: row.coordinator.issueId, + kind: "active_run_watchdog", + ownerType: "agent", + ownerAgentId: row.run.agentId, + cause: failureCode, + fingerprint: createHash("sha256").update(`${row.run.id}:${failureCode}`).digest("hex"), + evidence: { runId: row.run.id, hasEnvelope: !!persistedInput, hasCheckpoint: !!checkpoint }, + nextAction: "Inspect the interrupted native session and explicitly choose recovery; do not open a duplicate provider session.", + wakePolicy: null, + maxAttempts: 3, + }); + return false; + } + + const leaseExpiresAt = new Date(now.getTime() + 20 * 60_000); + const updated = await tx.update(nativeRunFinalizations).set({ + phase: "observed", + leaseOwner, + leaseExpiresAt, + failureCode: null, + failureDetail: null, + nextAttemptAt: null, + updatedAt: now, + }).where(and( + eq(nativeRunFinalizations.runId, row.run.id), + eq(nativeRunFinalizations.phase, row.coordinator.phase), + )).returning({ runId: nativeRunFinalizations.runId }).then((rows) => rows[0] ?? null); + if (!updated) return false; + await tx.update(heartbeatRuns).set({ + status: "running", + finishedAt: null, + error: null, + errorCode: null, + nativePhase: "observed", + nativePhaseUpdatedAt: now, + updatedAt: now, + }).where(eq(heartbeatRuns.id, row.run.id)); + return true; + }); + if (claimed) claims.push({ runId: candidate.runId, leaseOwner }); + } + return claims; +} /** Recovery is keyed only by persisted mode/coordinator state, never the live flag. */ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) { diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index deda18feb2a..576c23a7d61 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -1,12 +1,134 @@ +import type { Db } from "@paperclipai/db"; +import type { NativeInteractionResponseEnvelope } from "@paperclipai/paperclip-runner"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; + +export class NativeInteractionBridgeError extends Error { + constructor(readonly code: string, message: string) { + super(message); + this.name = "NativeInteractionBridgeError"; + } +} + /** - * Phase 6 never auto-approves model requests. Unsupported native requests are - * declined explicitly; typed response materialization stays in existing - * interaction/approval services. + * Materialize only interactions that authorized this run's wake. Resolution + * remains owned by the existing interaction service; this bridge is a + * read-only, identity-bound projection and never receives credentials. */ +export async function materializeNativeInteractionResponses(input: { + db: Db; + companyId: string; + issueId: string; + runId: string; + agentId: string; + interactionIds: string[]; +}): Promise { + const requestedIds = new Set(input.interactionIds.filter(Boolean)); + if (requestedIds.size === 0) return []; + const interactions = await issueThreadInteractionService(input.db).listForIssue(input.issueId); + const responses: NativeInteractionResponseEnvelope[] = []; + + for (const interaction of interactions) { + if (!requestedIds.has(interaction.id)) continue; + if (interaction.companyId !== input.companyId || interaction.issueId !== input.issueId) { + throw new NativeInteractionBridgeError( + "native_interaction_binding_mismatch", + `Interaction ${interaction.id} is not bound to the native company and issue`, + ); + } + if ( + interaction.resolvedByRunId === input.runId + || ( + interaction.resolvedByAgentId === input.agentId + && interaction.createdByAgentId === input.agentId + ) + ) { + throw new NativeInteractionBridgeError( + "native_interaction_self_approval", + `Run ${input.runId} cannot consume a response it resolved itself`, + ); + } + if (!interaction.resolvedByUserId && !interaction.resolvedByAgentId) { + throw new NativeInteractionBridgeError( + "native_interaction_unresolved", + `Interaction ${interaction.id} has no authorized resolver`, + ); + } + + if (interaction.kind === "request_confirmation") { + if (interaction.resolvedByAgentId === input.agentId) { + throw new NativeInteractionBridgeError( + "native_interaction_self_approval", + `Agent ${input.agentId} cannot consume a confirmation it resolved`, + ); + } + if (interaction.payload.toolAction !== undefined) { + throw new NativeInteractionBridgeError( + "native_interaction_governed_request_unsupported", + "Governed tool-action confirmations cannot enter a native model envelope", + ); + } + if ( + (interaction.status !== "accepted" && interaction.status !== "rejected") + || !interaction.result + || (interaction.result.outcome !== "accepted" && interaction.result.outcome !== "rejected") + ) { + throw new NativeInteractionBridgeError( + "native_interaction_unresolved", + `Confirmation ${interaction.id} is not authoritatively resolved`, + ); + } + responses.push({ + interactionId: interaction.id, + kind: interaction.kind, + response: { + status: interaction.status, + result: structuredClone(interaction.result) as unknown as Record, + }, + }); + continue; + } + + if (interaction.kind === "ask_user_questions") { + if (interaction.status !== "answered" || !interaction.result || interaction.result.cancelled === true) { + throw new NativeInteractionBridgeError( + "native_interaction_unresolved", + `Question interaction ${interaction.id} is not authoritatively answered`, + ); + } + responses.push({ + interactionId: interaction.id, + kind: interaction.kind, + response: { + status: interaction.status, + result: structuredClone(interaction.result) as unknown as Record, + }, + }); + continue; + } + + throw new NativeInteractionBridgeError( + "native_runtime_request_unsupported", + `Interaction kind ${interaction.kind} is not supported by the native input contract`, + ); + } + + const missing = [...requestedIds].filter((id) => !responses.some((response) => response.interactionId === id)); + if (missing.length > 0) { + throw new NativeInteractionBridgeError( + "native_interaction_missing", + `Native interaction response not found: ${missing.join(", ")}`, + ); + } + return responses.sort((left, right) => left.interactionId.localeCompare(right.interactionId)); +} + +/** Native runtimes never receive credentials and never auto-approve requests. */ export function rejectUnsupportedNativeRuntimeRequest(requestKind: string) { return { accepted: false as const, code: "native_runtime_request_unsupported" as const, requestKind, + credentialsInjected: false as const, + selfApproval: false as const, }; } diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 7d1f1a8d719..7556ac08abe 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -13,7 +13,11 @@ vi.mock("@paperclipai/paperclip-runner", () => ({ executeNativeSession: state.execute, })); -import { cancelNativeSession, executePaperclipNativeSession } from "./native-session-executor.js"; +import { + cancelNativeSession, + executePaperclipNativeSession, + nativeSessionFailureDisposition, +} from "./native-session-executor.js"; const execution = { binding: { @@ -116,3 +120,24 @@ describe("native session cancellation", () => { await running; }); }); + +describe("native session bounded recovery", () => { + it("retries the same run twice and stops at the third failed attempt", () => { + const now = new Date("2026-08-09T00:00:00.000Z"); + expect(nativeSessionFailureDisposition(1, now)).toEqual({ + phase: "retryable_failure", + failureCode: "native_session_interrupted", + nextAttemptAt: new Date("2026-08-09T00:00:30.000Z"), + }); + expect(nativeSessionFailureDisposition(2, now)).toEqual({ + phase: "retryable_failure", + failureCode: "native_session_interrupted", + nextAttemptAt: new Date("2026-08-09T00:00:30.000Z"), + }); + expect(nativeSessionFailureDisposition(3, now)).toEqual({ + phase: "terminal_failure", + failureCode: "native_session_retry_exhausted", + nextAttemptAt: null, + }); + }); +}); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 426ed3cb7d0..3deb6d7cbbd 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import type { AdapterExecutionResult } from "../../adapters/index.js"; import type { NativeFinalizationResult } from "@paperclipai/shared"; import type { NativeExecutionInputV1, NativeSession } from "@paperclipai/paperclip-runner"; @@ -10,6 +10,7 @@ import type { Db } from "@paperclipai/db"; import { and, eq } from "drizzle-orm"; import { heartbeatRuns, nativeRunFinalizations } from "@paperclipai/db"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { issueRecoveryActionService } from "../issue-recovery-actions.js"; type ActiveNativeSession = { session: NativeSession; @@ -18,6 +19,15 @@ type ActiveNativeSession = { const activeNativeSessions = new Map(); +export function nativeSessionFailureDisposition(attempt: number, now = new Date()) { + const exhausted = attempt >= 3; + return { + phase: exhausted ? "terminal_failure" as const : "retryable_failure" as const, + failureCode: exhausted ? "native_session_retry_exhausted" as const : "native_session_interrupted" as const, + nextAttemptAt: exhausted ? null : new Date(now.getTime() + 30_000), + }; +} + export async function cancelNativeSession(runId: string, reason: string): Promise { const active = activeNativeSessions.get(runId); if (!active) return false; @@ -37,11 +47,12 @@ export async function executePaperclipNativeSession(input: { db: Db; execution: NativeExecutionInputV1; runnerInstanceId: string; + leaseOwner?: string; }): Promise { - const leaseOwner = `${input.runnerInstanceId}:${randomUUID()}`; + const leaseOwner = input.leaseOwner ?? `${input.runnerInstanceId}:${randomUUID()}`; const leaseNow = new Date(); const leaseExpiresAt = new Date(leaseNow.getTime() + 20 * 60_000); - await input.db.transaction(async (tx) => { + const attempt = await input.db.transaction(async (tx) => { const coordinator = await tx.select().from(nativeRunFinalizations) .where(and( eq(nativeRunFinalizations.runId, input.execution.binding.runId), @@ -71,6 +82,7 @@ export async function executePaperclipNativeSession(input: { nativePhaseUpdatedAt: leaseNow, updatedAt: leaseNow, }).where(eq(heartbeatRuns.id, coordinator.runId)); + return coordinator.attempt + 1; }); const controlPlaneInstanceId = `${input.runnerInstanceId}:control`; const controlPlane = new PaperclipControlPlanePort(input.db, { @@ -101,22 +113,57 @@ export async function executePaperclipNativeSession(input: { }); } catch (error) { const now = new Date(); - await input.db.update(nativeRunFinalizations).set({ - phase: "retryable_failure", - leaseOwner: null, - leaseExpiresAt: null, - failureCode: "native_session_interrupted", - failureDetail: { - message: error instanceof Error ? error.message.slice(0, 2_000) : String(error).slice(0, 2_000), - recoveryOwner: { kind: "agent", agentId: input.execution.binding.agentId }, - nextAction: "Resume the persisted native provider session after the retry delay.", - }, - nextAttemptAt: new Date(now.getTime() + 30_000), - updatedAt: now, - }).where(and( - eq(nativeRunFinalizations.runId, input.execution.binding.runId), - eq(nativeRunFinalizations.leaseOwner, leaseOwner), - )); + const { phase, failureCode, nextAttemptAt } = nativeSessionFailureDisposition(attempt, now); + const exhausted = phase === "terminal_failure"; + const message = error instanceof Error ? error.message.slice(0, 2_000) : String(error).slice(0, 2_000); + await input.db.transaction(async (tx) => { + const updated = await tx.update(nativeRunFinalizations).set({ + phase, + leaseOwner: null, + leaseExpiresAt: null, + failureCode, + failureDetail: { + message, + originalFailureCode: "native_session_interrupted", + recoveryOwner: { kind: "agent", agentId: input.execution.binding.agentId }, + nextAction: exhausted + ? "Inspect the persisted native session after its bounded resume budget was exhausted." + : "Resume this same run from its persisted native provider checkpoint after the retry delay.", + }, + nextAttemptAt, + updatedAt: now, + }).where(and( + eq(nativeRunFinalizations.runId, input.execution.binding.runId), + eq(nativeRunFinalizations.leaseOwner, leaseOwner), + )).returning({ runId: nativeRunFinalizations.runId }).then((rows) => rows[0] ?? null); + if (!updated) throw new Error("native_session_lease_lost"); + await tx.update(heartbeatRuns).set({ + nativePhase: phase, + nativePhaseUpdatedAt: now, + error: message, + errorCode: failureCode, + updatedAt: now, + }).where(eq(heartbeatRuns.id, input.execution.binding.runId)); + await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({ + companyId: input.execution.binding.companyId, + sourceIssueId: input.execution.binding.issueId, + kind: "active_run_watchdog", + ownerType: "agent", + ownerAgentId: input.execution.binding.agentId, + cause: failureCode, + fingerprint: createHash("sha256") + .update(`${input.execution.binding.runId}:${failureCode}`) + .digest("hex"), + evidence: { runId: input.execution.binding.runId, coordinatorAttempt: attempt }, + nextAction: exhausted + ? "Inspect or explicitly restart the exhausted persisted native session." + : "Resume the persisted native session on the same heartbeat run.", + wakePolicy: nextAttemptAt + ? { kind: "resume_native_run", runId: input.execution.binding.runId, notBefore: nextAttemptAt.toISOString() } + : null, + maxAttempts: 3, + }); + }); throw error; } await input.db.update(nativeRunFinalizations).set({ 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 71b2bd6ae1c..100ce4d925d 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 @@ -272,6 +272,11 @@ describe("PaperclipControlPlanePort conformance", () => { highestContiguousSourceSeq: 3, duplicateDisposition: "duplicate", terminalReplayIdempotent: true, + openBindingRejected: true, + eventIdMutationRejected: true, + eventSequenceMutationRejected: true, + replayBindingRejected: true, + resultMutationRejected: true, }); await expect(db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).resolves.toHaveLength(1); await finalizeNativeRun({