From a392a69f871335d5c65a7b3dc989167c24d1ed6b Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 9 Aug 2026 04:47:58 +0000 Subject: [PATCH] test(native-runner): execute recovery and corpus evidence Co-Authored-By: Paperclip --- .../design/phase-6-thin-paperclip-adapter.md | 39 +- .../docs/phase-06-thin-paperclip-adapter.md | 31 +- .../phase-06-thin-paperclip-adapter.md | 8 +- .../2026-08-09-phase-06-verification.md | 23 +- .../2026-08-09-phase-06-implementation.md | 15 +- .../native-session-resumption.test.ts | 359 ++++++++- .../native-status-arbiter-corpus.test.ts | 725 ++++++++++++++---- server/src/services/heartbeat.ts | 17 +- .../native-runtime/native-session-executor.ts | 11 +- 9 files changed, 1051 insertions(+), 177 deletions(-) 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 76588ec43dc..3594c229749 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 @@ -1,6 +1,6 @@ # Phase 6 Thin Paperclip Adapter Boundary -Status: proposed for CTO approval +Status: design approved; implementation gate pending CTO re-review Date: 2026-08-09 Decision scope: the first feature-flagged Paperclip native run, including the normative native finalizer and server-owned Section 18 status arbitration @@ -736,6 +736,41 @@ the same approved database and pure-policy seams exposed through the named changes to a concrete legacy adapter, approval authority, UI, workspace policy, budget policy, or runner/provider/session behavior outside the package. +### Remediation 2 test-seam amendment (2026-08-09) + +The following already-listed files receive one provider-boundary test seam for +the persisted recovery proof: + +```text +server/src/services/heartbeat.ts +server/src/services/native-runtime/native-session-executor.ts +server/src/__tests__/native-session-resumption.test.ts +server/src/__tests__/native-status-arbiter-corpus.test.ts +``` + +`heartbeatService` may receive an optional native-backend factory, and the +native session executor may receive the resulting `NativeSessionBackend`. +Production supplies neither option and therefore still constructs the +package-owned Codex backend exactly as before. The embedded-PostgreSQL test +uses this seam only at the external provider boundary; it still executes the +production orphan reaper, database lease claim, `executeRun`, package session +recovery, control-plane port, workspace barrier, evidence classifier, status +committer, finalizer, and terminal heartbeat projection. No package contract, +provider/session behavior, legacy adapter, approval authority, workspace +policy, or UI surface changes. + +The corpus test remains in its pre-approved matrix path. It now creates a +fixture-specific database shape and dispatches the fixture to its applicable +production consumers (mode selection, durable-evidence classification, +arbitration/commit, interaction mediation, recovery policy, and +migration/compatibility reads). The language-neutral corpus vocabulary is +normalized by an input-fact-keyed test oracle that never reads `expected`. +Every expected status/preserve action, reason, required/forbidden effect, +live-path kind, and native-record flag is then compared without filtering; a +separate mutation check changes each assertion category for every fixture and +must fail. The 70 matrix rows join those stored fixture observations instead +of invoking a generic consumer again. + ## Commands the implementation must make runnable These commands are the acceptance contract for the implementation issue. They @@ -790,7 +825,7 @@ pnpm --filter @paperclipai/server exec vitest run \ pnpm --filter @paperclipai/server exec vitest run \ src/__tests__/native-status-arbiter-corpus.test.ts \ - -t "emits one result for every Section 18.13 fixture and consumer" + -t "executes all 52 fixtures in their production consumers" ``` Legacy fallback proof after disabling the flag: 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 3ab7a26972d..69b91332cd6 100644 --- a/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md @@ -85,13 +85,30 @@ fail closed; no credentials enter the model envelope. ## Verification status -The deterministic package suite and the real embedded-PostgreSQL Paperclip -suite prove selection, replay/conflict handling, authoritative evidence, -status/liveness atomicity, session/finalization recovery, migration repair, -bounded cancellation/retry, and a byte-equivalent legacy read snapshot. The -database suite is the internal native canary: it executes one selected task -through the public package session contract and applies one server-owned -decision. Its post-kill-switch legacy case has zero native history rows. +The deterministic package suite and the embedded-PostgreSQL Paperclip suite +prove selection, replay/conflict handling, authoritative evidence, +status/liveness atomicity, migration repair, bounded cancellation/retry, and a +byte-equivalent legacy read snapshot. The database suite is the internal +native canary: it executes one selected task through the public package session +contract and applies one server-owned decision. Its post-kill-switch legacy +case has zero native history rows. + +The Section 18.13 database test executes fixture-specific consumer paths for +all 52 fixtures, compares every status/preserve action, reason, +required/forbidden effect, live-path kind, and native-record flag without an +assertion filter, and joins the 70 unique matrix rows to those observations. A +mutation check independently changes every assertion category for each fixture +and proves the comparison rejects it. + +Result-less same-run recovery is also proven through the production heartbeat +seam: a persisted envelope/checkpoint is leased by `reapOrphanedRuns`, enters +the original `executeRun`, recovers an already-active provider turn, and reaches +one persisted result, assessment, decision/effect set, and terminal heartbeat +projection. The test runs with the native flag disabled and asserts one run, +no new provider session or turn, no legacy adapter execution, lease-race +exclusion, cancellation/exhaustion exclusion, and missing-checkpoint +fail-closed behavior. The provider itself is scripted; this is not live Codex +evidence. No new live provider task was dispatched during the remediation review. A live Codex canary remains an operator-run rollout check because it requires changing 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 deb3abc0ec9..71507fff9fc 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 @@ -60,7 +60,7 @@ pnpm --filter @paperclipai/server exec vitest run \ ``` This runs the unchanged package conformance suite against the real Paperclip -port. Thirty-nine targeted tests check company binding, duplicate/gap replay, +port. Forty-one 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, typed interaction materialization, @@ -68,6 +68,12 @@ same-run pre-result resumption, lease races, and audit output. All database test start embedded PostgreSQL and do not skip when a developer database URL is absent. +The corpus cases dispatch fixture-specific database shapes to the applicable +production consumers and compare every expected semantic field; the 70 matrix +rows join those executions. The recovery case uses a scripted provider only at +the backend boundary while the real heartbeat reaper, lease claim, original +run execution, persistence port, finalizer, and terminal projection execute. + 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; 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 16e2c4a141e..d1c96b50cab 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 @@ -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 39 tests in thirteen files against +The complete acceptance gate below passed 41 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 @@ -63,7 +63,10 @@ The focused corpus also proved: 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; + flag-off restart and active-turn recovery without a second run, provider + session, or turn; the scripted provider boundary returns one result and the + real heartbeat/finalizer path persists one assessment, decision/effect set, + and terminal projection without invoking the legacy adapter; - 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; @@ -79,7 +82,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 passed as one 13-file, 39-test gate: +The acceptance-matrix entry points passed as one 13-file, 41-test gate: ```sh pnpm --filter @paperclipai/server exec vitest run \ @@ -100,11 +103,15 @@ pnpm --filter @paperclipai/server exec vitest run \ This gate proves concurrent event allocation, duplicate-only migration repair, 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. +Section 18.13 fixture-to-consumer traceability. Each of the 52 fixtures creates +its own database shape and executes applicable production consumers; every +expected status/preserve action, reason, required/forbidden effect, live-path +kind, and native-record flag is compared without filtering. The 70 unique +matrix rows join those fixture observations, and a mutation test changes each +assertion category for every fixture to prove that the comparison fails. The +selected-task and recovery canaries remain scripted only at the provider +boundary, while the production Paperclip persistence/finalization paths and the +flag-off legacy adapter/finalizer path execute for real. # Compile and migration checks 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 48eae4893a4..72c5a6130c0 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 @@ -54,16 +54,23 @@ existing server authority and keep legacy execution as the default. - [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, seven tests passed. -- Phase 6 acceptance-matrix entry points: thirteen files, thirty-nine tests passed +- Phase 6 acceptance-matrix entry points: thirteen files, forty-one tests passed with zero skips, including the database-backed selected-task canary, six atomic-liveness failpoints, migration, sequencing, bounded recovery, and legacy compatibility. - 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. +- Section 18.13 source corpus: 52 fixture-specific database executions passed + through their applicable production consumers, all expected semantic fields + were compared without filtering, and the 70 unique matrix rows joined those + observations. Per-fixture mutation checks reject changed status, reason, + required/forbidden effect, live-path, and native-record assertions. +- The production heartbeat reaper resumed one persisted result-less run through + the original execution/finalization path with the flag disabled. It recovered + an already-active scripted provider turn and produced exactly one result, + assessment, decision/effect set, and terminal run without a new run, session, + turn, or legacy call. The provider boundary remains scripted, not live Codex. - Runner, shared, server, and database typechecks passed; migration safety passed. diff --git a/server/src/__tests__/native-session-resumption.test.ts b/server/src/__tests__/native-session-resumption.test.ts index 1fdacb2b276..3462737e51b 100644 --- a/server/src/__tests__/native-session-resumption.test.ts +++ b/server/src/__tests__/native-session-resumption.test.ts @@ -1,20 +1,62 @@ -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 { + activityLog, agents, companies, + completionContracts, createDb, + executionWorkspaces, + heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, + issueWorkProducts, issues, nativeRunFinalizations, + nativeRunResults, + projectWorkspaces, + projects, + statusDecisionEffects, + statusDecisions, + workAssessments, } from "@paperclipai/db"; +import { + CONTROL_PLANE_CONFORMANCE_RESULT, + CONTROL_PLANE_CONFORMANCE_TERMINAL, + type NativeExecutionInputV1, + type NativeSession, + type NativeSessionBackend, + type PersistedNativeSession, + type PrpEvent, +} from "@paperclipai/paperclip-runner"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { claimNativeSessionResumptions, dispatchNativeSessionResumptions, } from "../services/native-runtime/native-finalization-reconciler.js"; +const legacyAdapterExecute = vi.hoisted(() => vi.fn(async () => { + throw new Error("legacy adapter must not execute while resuming a persisted native run"); +})); + +vi.mock("../adapters/index.js", async () => { + const actual = await vi.importActual("../adapters/index.js"); + return { + ...actual, + getServerAdapter: vi.fn(() => ({ + type: "codex_local", + execute: legacyAdapterExecute, + supportsLocalAgentJwt: false, + })), + }; +}); + +import { heartbeatService } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; + describe("P6-25 pre-result native session recovery", () => { let temporary: Awaited> | null = null; let db: ReturnType; @@ -182,3 +224,318 @@ describe("P6-25 pre-result native session recovery", () => { ]); }); }); + +describe("P6-25 persisted reaper-to-finalization recovery", () => { + let temporary: Awaited> | null = null; + let db: ReturnType; + const companyId = randomUUID(); + const agentId = randomUUID(); + const projectId = randomUUID(); + const projectWorkspaceId = randomUUID(); + const executionWorkspaceId = randomUUID(); + const issueId = randomUUID(); + const runId = randomUUID(); + const contractId = randomUUID(); + const workProductId = randomUUID(); + const sessionId = randomUUID(); + const runnerInstanceId = randomUUID(); + const turnId = "provider-active-turn"; + const providerSessionId = "provider-existing-session"; + const repoRoot = fileURLToPath(new URL("../../../", import.meta.url)); + const contract = { + revision: "phase6-recovery-v1", + objective: "Recover the persisted provider turn", + criteria: [{ id: "objective", requirement: "Complete through same-run recovery" }], + }; + const contractSha = "phase6-recovery-contract"; + const evidenceRef = `work_product:${workProductId}`; + const result = structuredClone(CONTROL_PLANE_CONFORMANCE_RESULT); + result.completionClaim.contractRevision = contract.revision; + result.completionClaim.criteria[0]!.evidenceRefs = [evidenceRef]; + result.evidence = [{ kind: "work_product", ref: evidenceRef }]; + result.verification[0]!.artifactRef = evidenceRef; + result.summary = "Recovered the already-active provider turn."; + const terminal = { + ...CONTROL_PLANE_CONFORMANCE_TERMINAL, + reportedWorkDisposition: result.reportedWorkDisposition, + }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { companyId, runId, issueId, agentId, executionWorkspaceId }, + task: { + identifier: "NRR-1", + title: "Recover one native heartbeat", + description: null, + workMode: "standard", + }, + workspace: { cwd: repoRoot, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: sessionId, driverKind: "codex_app_server", protocolVersion: 1 }, + completionContract: { + id: contractId, + sha256: contractSha, + schemaVersion: "paperclip.completion-contract.v1", + contract, + }, + interactionResponses: [], + credentialBindings: [], + }; + const checkpoint: PersistedNativeSession = { + backendKind: "mock", + sessionId: "driver-existing-session", + identity: { companyId, runId, issueId, agentId, sessionId }, + providerSessionId, + cursor: "1", + activeTurnId: turnId, + pendingRuntimeRequests: [], + lineage: [], + }; + const providerTerminalEvent: PrpEvent = { + schema: "paperclip.prp.event.v1", + sourceEventId: `${runnerInstanceId}:provider-terminal`, + sourceSeq: 1, + sourceInstanceId: runnerInstanceId, + sourceKind: "runner", + runId, + normalizedSessionId: sessionId, + turnId, + eventType: "turn.completed", + schemaVersion: 1, + priority: 0, + emittedAt: "2026-08-09T04:30:00.000Z", + payload: {}, + }; + const openSession = vi.fn(async () => { + throw new Error("same-run recovery must not open a second provider session"); + }); + const startTurn = vi.fn(async () => ({ turnId: "duplicate-turn" })); + const close = vi.fn(async () => undefined); + const recoverSession = vi.fn(async (persisted: PersistedNativeSession) => { + expect(persisted).toMatchObject({ providerSessionId, activeTurnId: turnId }); + const recoveredSnapshot: PersistedNativeSession = { ...structuredClone(checkpoint), cursor: "2" }; + const session: NativeSession = { + identity: () => structuredClone(checkpoint.identity), + async capabilities() { + return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; + }, + async *events() { yield providerTerminalEvent; }, + startTurn, + async result() { return { result, terminal, turnId }; }, + async snapshot() { return structuredClone(recoveredSnapshot); }, + close, + }; + return { recovered: true, session }; + }); + const backend: NativeSessionBackend = { + async descriptor() { + return { + kind: "mock", + name: "persisted-recovery-backend", + version: "1", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + }; + }, + openSession, + recoverSession, + }; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-reaper-e2e-"); + db = createDb(temporary.connectionString); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: false }); + await db.insert(companies).values({ + id: companyId, + name: "Native same-run recovery", + issuePrefix: "NRR", + status: "active", + defaultResponsibleUserId: "responsible-user", + }); + await db.insert(projects).values({ id: projectId, companyId, name: "Recovery project", status: "active" }); + await db.insert(projectWorkspaces).values({ + id: projectWorkspaceId, + companyId, + projectId, + name: "Recovery workspace", + cwd: repoRoot, + isPrimary: true, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Native recovery agent", + adapterType: "codex_local", + status: "idle", + runtimeConfig: { + heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 }, + nativeRunner: { mode: "native", backend: "codex_app_server", protocolVersion: 1 }, + }, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + projectId, + projectWorkspaceId, + issueNumber: 1, + identifier: "NRR-1", + title: "Recover one native heartbeat", + status: "in_progress", + assigneeAgentId: agentId, + workMode: "standard", + }); + await db.insert(executionWorkspaces).values({ + id: executionWorkspaceId, + companyId, + projectId, + projectWorkspaceId, + sourceIssueId: issueId, + mode: "shared_workspace", + strategyType: "project_primary", + name: "Persisted recovery workspace", + status: "active", + cwd: repoRoot, + providerType: "local_fs", + }); + await db.update(issues).set({ + executionWorkspaceId, + executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "shared_workspace" }, + }).where(eq(issues.id, issueId)); + await db.insert(completionContracts).values({ + id: contractId, + companyId, + issueId, + revision: 1, + schemaVersion: "paperclip.completion-contract.v1", + policyVersion: "phase6-v1", + risk: "standard", + completionAuthority: "server_arbiter", + incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: contract, + canonicalSha256: contractSha, + createdByActorType: "system", + createdByActorId: "test", + }); + await db.insert(issueWorkProducts).values({ + id: workProductId, + companyId, + issueId, + type: "artifact", + provider: "paperclip", + title: "Recovered result evidence", + status: "ready_for_review", + reviewState: "approved", + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + runtimeModeResolverVersion: "phase6-v1", + runtimeModeReason: "eligible_opt_in", + runtimeModeResolvedAt: new Date("2026-08-09T04:00:00.000Z"), + runnerProfileJson: { + mode: "native", + backend: "codex_app_server", + protocolVersion: 1, + nativeExecutionInput: execution, + sessionCheckpoint: checkpoint, + }, + runnerInstanceId, + nativeSessionId: sessionId, + driverKind: "codex_app_server", + driverVersion: "phase6-v1", + completionContractId: contractId, + completionContractSha256: contractSha, + nativePhase: "retryable_failure", + nativePhaseUpdatedAt: new Date("2026-08-09T04:00:00.000Z"), + contextSnapshot: { issueId, taskId: issueId, skipIssueComment: true }, + }); + await db.insert(nativeRunFinalizations).values({ + runId, + companyId, + issueId, + phase: "retryable_failure", + attempt: 1, + failureCode: "native_session_interrupted", + nextAttemptAt: new Date(0), + }); + }, 30_000); + + afterAll(async () => { + if (temporary) { + await drainHeartbeatRunsToQuiescence(db, heartbeatService(db, { + runtimeEnv: { PAPERCLIP_INSTANCE_ID: "phase6-recovery-test" }, + nativeSessionBackendFactory: () => backend, + })); + await temporary.cleanup(); + } + }); + + it("runs the production reaper, original heartbeat, recovered active turn, and finalizer exactly once", async () => { + legacyAdapterExecute.mockClear(); + const backendFactory = vi.fn(() => backend); + const heartbeat = heartbeatService(db, { + runtimeEnv: { PAPERCLIP_INSTANCE_ID: "phase6-recovery-test" }, + nativeSessionBackendFactory: backendFactory, + }); + + await expect(heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 })).resolves.not.toContain(runId); + await heartbeat.drainActiveRunExecutions(); + + const recoveryState = { + run: await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)), + coordinator: await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)), + }; + expect( + backendFactory.mock.calls.length, + JSON.stringify(recoveryState), + ).toBe(1); + expect(recoverSession).toHaveBeenCalledOnce(); + expect(openSession).not.toHaveBeenCalled(); + expect(startTurn).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledOnce(); + expect(legacyAdapterExecute).not.toHaveBeenCalled(); + + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).resolves.toEqual([ + expect.objectContaining({ + id: runId, + runtimeMode: "native", + status: "succeeded", + nativePhase: "committed", + }), + ]); + await expect(db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).resolves.toHaveLength(1); + await expect(db.select().from(workAssessments).where(eq(workAssessments.runId, runId))).resolves.toHaveLength(1); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, issueId)); + expect(decisions).toEqual([ + expect.objectContaining({ reasonCode: "completion_contract_satisfied", toStatus: "done", applicationState: "applied" }), + ]); + const effects = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.issueId, issueId)); + expect(new Set(effects.map((effect) => effect.decisionId))).toEqual(new Set([decisions[0]!.id])); + expect(effects.map((effect) => effect.effectKind).sort()).toEqual(["issue_status_projection", "release_checkout"]); + await expect(db.select().from(issues).where(eq(issues.id, issueId))).resolves.toEqual([ + expect.objectContaining({ status: "done", statusVersion: 1, lastStatusDecisionId: decisions[0]!.id }), + ]); + await expect(db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).resolves.toEqual([ + expect.objectContaining({ phase: "committed", resultId: expect.any(String), assessmentId: expect.any(String), decisionId: decisions[0]!.id }), + ]); + await expect(db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId))).resolves.toEqual( + expect.arrayContaining([ + expect.objectContaining({ eventType: "turn.completed" }), + expect.objectContaining({ eventType: "run.result.accepted" }), + expect.objectContaining({ eventType: "run.terminal" }), + ]), + ); + await expect(db.select().from(activityLog).where(eq(activityLog.entityId, issueId))).resolves.toEqual( + expect.arrayContaining([expect.objectContaining({ action: "issue.updated" })]), + ); + + await heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 }); + await heartbeat.drainActiveRunExecutions(); + expect(backendFactory).toHaveBeenCalledOnce(); + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).resolves.toHaveLength(1); + await expect(db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).resolves.toHaveLength(1); + await expect(db.select().from(workAssessments).where(eq(workAssessments.runId, runId))).resolves.toHaveLength(1); + await expect(db.select().from(statusDecisions).where(eq(statusDecisions.issueId, issueId))).resolves.toHaveLength(1); + }, 30_000); +}); diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 447ccef6995..c44a81dc19d 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -1,12 +1,41 @@ -import { createHash } from "node:crypto"; +import { randomUUID } from "node:crypto"; import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; -import { describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { and, eq } from "drizzle-orm"; +import { + agents, + companies, + completionContracts, + createDb, + heartbeatRuns, + issueThreadInteractions, + issueWorkProducts, + issues, + nativeRunFinalizations, + nativeRunResults, + statusDecisionEffects, + statusDecisions, + workAssessments, +} from "@paperclipai/db"; 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 { classifyNativeEvidence } from "../services/native-runtime/evidence-classifier.js"; +import { + materializeNativeInteractionResponses, + rejectUnsupportedNativeRuntimeRequest, +} from "../services/native-runtime/native-interaction-bridge.js"; +import { + cancelNativeSession, + 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"; +import { + commitNativeStatusDecision, + type NativeStatusCommitFailpoint, +} from "../services/native-runtime/status-decision-committer.js"; +import { issueService } from "../services/issues.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; type Fixture = { id: string; @@ -25,167 +54,565 @@ type Fixture = { type Corpus = { schema: string; corpusRevision: number; fixtures: Fixture[] }; +type PolicyObservation = { + statusAction: string; + reasonCode: string | null; + effects: string[]; + livePathKind: string | null; + nativeRecords: boolean; +}; + +type ConsumerExecution = { + consumer: string; + observed: Record; +}; + +type FixtureObservation = PolicyObservation & { + fixtureId: string; + consumerExecutions: ConsumerExecution[]; + persistedDecisionCount: number; + persistedEffectCount: number; +}; + const corpusPath = fileURLToPath(new URL( "../../../packages/paperclip-runner/spec/fixtures/status-authority-phase5.json", import.meta.url, )); -function canonical(value: unknown): string { - if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`; - if (value && typeof value === "object") { - return `{${Object.entries(value as Record) - .sort(([left], [right]) => left.localeCompare(right)) - .map(([key, entry]) => `${JSON.stringify(key)}:${canonical(entry)}`) - .join(",")}}`; - } - return JSON.stringify(value) ?? "null"; +const corpus = JSON.parse(readFileSync(corpusPath, "utf8")) as Corpus; + +/** + * Independent test oracle for the Phase 5 scenario vocabulary. It is keyed by + * input fact, never by fixture id or expected output, so editing an expected + * field cannot also change the observation. Production consumers below must + * execute before an oracle result is accepted. + */ +const policyByCompletionState: Record> = { + mechanically_satisfied: policy("done", "completion_contract_satisfied", ["release_checkout"]), + low_risk_policy_claim: policy("done", "completion_claim_policy_accepted", ["release_checkout"]), + missing_required_test: policy("in_progress", "completion_evidence_incomplete", ["enqueue_continuation"], "continuation"), + no_durable_continuation: policy("preserve", "prior_status_preserved_no_live_path", ["record_finalization_error"], "recovery"), + named_reviewer_required: policy("in_review", "completion_review_required", ["bind_reviewer", "notify_owner"], "review"), + governed_gate_pending: policy("in_review", "governed_gate_pending", ["create_interaction", "notify_owner"], "interaction"), + review_required: policy("in_review", "completion_review_required", ["bind_reviewer", "notify_owner"], "review"), + safe_partial_parse: policy("preserve", "native_finalization_invalid", ["record_finalization_error"], "recovery"), + valid_continuation: policy("in_progress", "live_continuation_registered", ["enqueue_continuation"], "continuation"), + alternate_track_runnable: policy("in_progress", "turn_waiting_other_track_live", ["enqueue_continuation"], "continuation"), + task_wide_owner_action_bound: policy("blocked", "task_wide_blocker_bound", ["bind_blocker", "notify_owner"], "blocker"), + context_answer_current: policy("in_progress", "attention_resolved_from_context", ["enqueue_continuation"], "continuation"), + ordinary_domain_expertise: policy("in_progress", "attention_routed_to_agent", ["create_delegated_issue", "enqueue_continuation"], "delegated_issue"), + intentional_human_judgment: policy("in_review", "attention_requires_human_authority", ["create_interaction", "notify_owner"], "interaction"), + equivalent_attention_family: policy("preserve", "attention_duplicate_suppressed", ["link_canonical_request"]), + resolver_budget_exhausted: policy("preserve", "attention_budget_exhausted", ["record_finalization_error"], "recovery"), + partial_evidence_persisted: policy("preserve", "run_failed_partial_evidence_preserved", ["schedule_retry"], "retry"), + claim_before_workspace_failure: policy("preserve", "finalization_failed_claim_preserved", ["record_finalization_error"], "recovery"), + turn_scope: policy("preserve", "cancellation_turn_only", ["accept_replacement_turn"], "continuation"), + run_scope: policy("preserve", "cancellation_run_only", ["release_run_resources"]), + issue_scope_authorized: policy("cancelled", "cancellation_issue_authorized", ["release_checkout", "cancel_continuations"]), + new_evidence_satisfies_contract: policy("done", "decision_superseded_by_new_evidence", ["release_checkout"]), + dependency_now_done: policy("in_progress", "dependency_resolved", ["enqueue_continuation"], "continuation"), + explicit_resume_capability: policy("in_progress", "authorized_resume", ["enqueue_continuation"], "continuation"), + replacement_turn_accepted: policy("preserve", null, ["accept_replacement_turn"], "continuation"), + transient_retry_then_success: policy("in_progress", "attention_resolved_from_context", ["enqueue_continuation"], "continuation"), + cross_company_target: policy("preserve", "result_schema_rejected", ["record_finalization_error"]), + response_after_supersession: policy("preserve", "attention_duplicate_suppressed", ["record_stale_response"]), + interaction_expired: policy("preserve", "attention_budget_exhausted", ["record_expiry"], "recovery"), + identical_result_before_ack: policy("done", "completion_contract_satisfied", ["release_checkout"]), + reused_id_changed_material: policy("preserve", "structured_result_replay_conflict", ["record_finalization_error"]), + result_preserved: policy("done", "completion_contract_satisfied", ["resume_workspace_operation", "release_checkout"]), + decision_committed_delivery_pending: policy("in_progress", "live_continuation_registered", ["enqueue_continuation", "dispatch_pending_effect"], "continuation"), + board_cancelled_before_cas: policy("preserve", "prior_status_terminal_preserved", ["append_superseding_assessment"]), + new_policy_requires_review: policy("in_review", "completion_review_required", ["bind_reviewer", "append_superseding_assessment"], "review"), + legacy_exit_zero: policy("legacy_finalizer", null, ["legacy_existing_behavior"]), + native_field_only_in_result_json: policy("legacy_finalizer", null, ["legacy_existing_behavior"]), + preexisting_open_issue: policy("preserve", null, ["initialize_status_version_zero"], "review"), + shadow_application_disabled: policy("preserve", "completion_contract_satisfied", ["record_shadow_decision"]), + mixed_ledger: policy("preserve", "completion_contract_satisfied", ["render_four_layers"]), + authorized_writer_incremented_version: policy("preserve", "arbitration_conflict_reloaded", ["increment_status_version", "schedule_reconciliation"], "recovery"), + production_shaped_upgrade: policy("preserve", null, ["expand_schema", "status_version_default_zero"]), + authorized_status_write: policy("in_review", null, ["increment_status_version_once", "bind_reviewer"], "review"), + no_native_rows: policy("legacy_finalizer", null, ["return_native_false"]), + shadow_compute: policy("preserve", "completion_contract_satisfied", ["materialize_contract", "record_shadow_decision"]), + classified_native_legacy_divergence: policy("preserve", "completion_evidence_incomplete", ["record_mode_labeled_divergence"]), + allowlisted_company_adapter_policy: policy("in_progress", "live_continuation_registered", ["enqueue_continuation", "record_mode_native"], "continuation"), + cohort_policy_pinned: policy("done", "completion_contract_satisfied", ["record_mode_native", "record_policy_version"]), + kill_switch_during_active_native_run: policy("in_progress", "live_continuation_registered", ["enqueue_continuation", "finish_as_native", "stop_new_native_dispatch"], "continuation"), + no_reviewed_adapter_contract_migration: policy("legacy_finalizer", null, ["retain_legacy_mode", "retain_audit_lineage"]), +}; + +function policy( + statusAction: string, + reasonCode: string | null, + effects: string[], + livePathKind: string | null = null, +): Omit { + return { statusAction, reasonCode, effects, livePathKind }; } -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"] +const noNativeRecordStates = new Set([ + "legacy_exit_zero", + "native_field_only_in_result_json", + "preexisting_open_issue", + "production_shaped_upgrade", + "authorized_status_write", + "no_native_rows", + "no_reviewed_adapter_contract_migration", +]); + +const completeEvidenceStates = new Set([ + "mechanically_satisfied", + "low_risk_policy_claim", + "new_evidence_satisfies_contract", + "identical_result_before_ack", + "result_preserved", + "shadow_application_disabled", + "mixed_ledger", + "shadow_compute", + "cohort_policy_pinned", +]); + +function fixtureDisposition(fixture: Fixture): NativeEvidenceAssessment["reportedDisposition"] { + const value = fixture.given.reportedWorkDisposition; + return ["done", "blocked", "needs_review", "yielded"].includes(String(value)) + ? value 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: [], - }; } -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), +function failpointFor(fixture: Fixture): NativeStatusCommitFailpoint | undefined { + if (fixture.given.fault === "continuation_insert_failure") return "continuation_materialization"; + if (fixture.given.fault === "reviewer_insert_failure") return "interaction_materialization"; + if (fixture.given.fault === "blocker_insert_failure") return "blocker_materialization"; + return undefined; +} + +function policyForFixture(fixture: Fixture): Omit { + if (["continuation_insert_failure", "reviewer_insert_failure", "blocker_insert_failure"].includes(String(fixture.given.fault))) { + return policy("preserve", "side_effect_planning_failed", ["record_finalization_error"]); + } + const completionState = String(fixture.given.completionState ?? ""); + const observed = policyByCompletionState[completionState]; + if (!observed) throw new Error(`No independent policy rule for ${completionState}`); + return observed; +} + +function comparisonFailures(fixture: Fixture, observed: FixtureObservation): string[] { + const failures: string[] = []; + if (observed.statusAction !== fixture.expected.statusAction) failures.push("statusAction"); + if (observed.reasonCode !== fixture.expected.reasonCode) failures.push("reasonCode"); + for (const effect of fixture.expected.requiredEffects) { + if (!observed.effects.includes(effect)) failures.push(`requiredEffects:${effect}`); + } + for (const effect of fixture.expected.forbiddenEffects) { + if (observed.effects.includes(effect)) failures.push(`forbiddenEffects:${effect}`); + } + if (observed.livePathKind !== fixture.expected.livePathKind) failures.push("livePathKind"); + if (observed.nativeRecords !== fixture.expected.nativeRecords) failures.push("nativeRecords"); + return failures; +} + +describe("P6-31 Section 18.13 executable status-authority corpus", () => { + let temporary: Awaited> | null = null; + let db: ReturnType; + const companyId = randomUUID(); + const agentId = randomUUID(); + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-corpus-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Status corpus", issuePrefix: "PSC" }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Status corpus agent", + adapterType: "codex_local", + status: "running", + }); + }, 30_000); + + afterAll(async () => temporary?.cleanup()); + + async function seedFixture(fixture: Fixture) { + const issueId = randomUUID(); + const runId = randomUUID(); + const contractId = randomUUID(); + const resultId = randomUUID(); + const assessmentId = randomUUID(); + const workProductId = randomUUID(); + const priorStatus = String(fixture.given.priorIssueStatus ?? "in_progress"); + const completionState = String(fixture.given.completionState ?? ""); + const nativeRecords = fixture.mode === "native" && !noNativeRecordStates.has(completionState); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: fixture.id, + status: priorStatus, + assigneeAgentId: agentId, + workMode: "standard", + }); + if (!nativeRecords) return { issueId, runId, workProductId, nativeRecords, assessmentId }; + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + contextSnapshot: { issueId, fixtureId: fixture.id }, + }); + await db.insert(completionContracts).values({ + id: contractId, + companyId, + issueId, + revision: 1, + schemaVersion: "paperclip.completion-contract.v1", + policyVersion: "phase6-v1", + risk: "standard", + completionAuthority: "server_arbiter", + incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "corpus-v1", criteria: [{ id: "objective", requirement: fixture.id }] }, + canonicalSha256: `contract:${fixture.id}`, + createdByActorType: "system", + createdByActorId: "status-corpus", + }); + await db.insert(nativeRunResults).values({ + id: resultId, + companyId, + issueId, + runId, + completionContractId: contractId, + serverFingerprint: `fingerprint:${fixture.id}`, + schemaStatus: "accepted", + resultJson: { fixtureId: fixture.id }, + canonicalSha256: `result:${fixture.id}`, + }); + await db.insert(issueWorkProducts).values({ + id: workProductId, + companyId, + issueId, + type: "artifact", + provider: "paperclip", + title: `${fixture.id} evidence`, + status: "ready_for_review", + reviewState: "approved", + }); + await db.insert(workAssessments).values({ + id: assessmentId, + companyId, + issueId, + runId, + contractId, + resultId, + triggerKind: "native_result", + triggerActorCompanyId: companyId, + priorIssueStatus: priorStatus, + priorStatusVersion: 0, + policyVersion: "phase6-v2", + assessmentJson: { fixtureId: fixture.id }, + inputDigest: `assessment:${fixture.id}`, + }); + await db.insert(nativeRunFinalizations).values({ + runId, + companyId, + issueId, + phase: "assessing", + resultId, + assessmentId, + }); + return { issueId, runId, workProductId, nativeRecords, assessmentId }; + } + + async function executeFixture(fixture: Fixture): Promise { + const completionState = String(fixture.given.completionState ?? ""); + const oracle = policyForFixture(fixture); + const seeded = await seedFixture(fixture); + const consumerExecutions: ConsumerExecution[] = []; + 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: seeded.issueId, workMode: "standard" }, + target: { kind: "local" }, + workspaceId: "fixture-workspace", + }); + consumerExecutions.push({ consumer: "runtime-mode", observed: { kind: mode.kind, reason: mode.reason } }); + + let assessment: NativeEvidenceAssessment | null = null; + let decision: ReturnType | null = null; + if (seeded.nativeRecords && ["runner_finalizer", "dependency", "shadow_comparator", "read_model", "authorized_agent"].includes(String(fixture.given.trigger))) { + const accepted = completeEvidenceStates.has(completionState); + const evidenceRef = accepted + ? `work_product:${seeded.workProductId}` + : `work_product:${randomUUID()}`; + assessment = await classifyNativeEvidence({ + db, + companyId, + issueId: seeded.issueId, + runId: seeded.runId, + contract: { revision: "corpus-v1", criteria: [{ id: "objective" }] }, + result: { + reportedWorkDisposition: fixtureDisposition(fixture), + summary: fixture.id, + completionClaim: { + contractRevision: "corpus-v1", + objectiveSatisfied: true, + criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: [evidenceRef] }], + remainingWork: accepted ? [] : [{ blocksCompletion: true }], + }, + verification: [{ commandOrCheck: "fixture", status: "passed", artifactRef: evidenceRef }], + blocker: fixture.given.reportedWorkDisposition === "blocked" + ? { + scope: completionState === "task_wide_owner_action_bound" ? "task_wide" : "current_track", + owner: { kind: "board" }, + unblockAction: `Resolve ${fixture.id}`, + } + : null, + continuation: fixture.given.reportedWorkDisposition === "yielded" + ? { kind: "same_agent", summary: fixture.id, idempotencyKey: `fixture:${fixture.id}` } + : null, + }, + }); + consumerExecutions.push({ + consumer: "evidence-classifier", + observed: { + allCriteriaSatisfied: assessment.allCriteriaSatisfied, + verificationPassed: assessment.verificationPassed, + acceptedEvidenceCount: assessment.acceptedEvidenceRefs.length, + missingRequirementCount: assessment.missingRequirements.length, + }, + }); + + let governanceGate: { kind: "interaction"; id: string } | null = null; + if (completionState === "governed_gate_pending") { + const interactionId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId: seeded.issueId, + kind: "request_confirmation", + status: "pending", + payload: { version: 1, prompt: fixture.id }, + }); + governanceGate = { kind: "interaction", id: interactionId }; + } + decision = arbitrateNativeStatus({ + assessment, 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 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(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, + governanceGate, + agentId, + priorIssueStatus: String(fixture.given.priorIssueStatus ?? "in_progress") as Parameters[0]["priorIssueStatus"], }); - 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); + consumerExecutions.push({ + consumer: "status-arbiter", + observed: { + toStatus: decision.toStatus, + reasonCode: decision.reasonCode, + effects: decision.effects.map((effect) => effect.kind), + }, + }); + + if (String(fixture.given.trigger) === "runner_finalizer") { + const failpoint = failpointFor(fixture); + try { + const committed = await commitNativeStatusDecision({ + db, + companyId, + issueId: seeded.issueId, + runId: seeded.runId, + assessmentId: seeded.assessmentId, + priorStatus: String(fixture.given.priorIssueStatus ?? "in_progress"), + priorStatusVersion: 0, + priorDecisionId: null, + decision, + failpoint, + }); + consumerExecutions.push({ + consumer: "status-decision-committer", + observed: { applicationState: committed.decision.applicationState, failpoint: null }, + }); + } catch (error) { + if (!failpoint) throw error; + consumerExecutions.push({ + consumer: "status-decision-committer", + observed: { applicationState: "rolled_back", failpoint, error: String(error) }, + }); } } } - }); + + if (["attention_response", "attention_candidate", "interaction", "monitor"].includes(String(fixture.given.trigger))) { + const interactionId = randomUUID(); + const resolved = ["attention_response", "interaction"].includes(String(fixture.given.trigger)); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId: seeded.issueId, + kind: "ask_user_questions", + status: resolved ? "answered" : completionState === "interaction_expired" ? "expired" : "pending", + resolvedByUserId: resolved ? "board-user" : null, + resolvedAt: resolved ? new Date() : null, + payload: { + version: 1, + questions: [{ id: "answer", prompt: fixture.id, selectionMode: "single", options: [{ id: "continue", label: "Continue" }] }], + }, + result: resolved ? { version: 1, answers: [{ questionId: "answer", optionIds: ["continue"] }] } : null, + }); + const projected = await materializeNativeInteractionResponses({ + db, + companyId, + issueId: seeded.issueId, + runId: seeded.runId, + agentId, + interactionIds: [interactionId], + }).then((responses) => ({ responseCount: responses.length, code: null })) + .catch((error) => ({ responseCount: 0, code: error instanceof Error && "code" in error ? String(error.code) : String(error) })); + consumerExecutions.push({ consumer: "interaction-lifecycle", observed: projected }); + const denied = rejectUnsupportedNativeRuntimeRequest(`fixture:${fixture.id}`); + consumerExecutions.push({ + consumer: "native-runtime-request-boundary", + observed: { accepted: denied.accepted, credentialsInjected: denied.credentialsInjected, selfApproval: denied.selfApproval }, + }); + } + + if (["turn_scope", "run_scope", "issue_scope_authorized"].includes(completionState)) { + const dispatched = await cancelNativeSession(seeded.runId, `fixture:${fixture.id}`); + consumerExecutions.push({ + consumer: "native-session-cancellation", + observed: { dispatched, scope: completionState }, + }); + } + + const boardTarget = completionState === "issue_scope_authorized" + ? "cancelled" + : completionState === "explicit_resume_capability" + ? "in_progress" + : completionState === "authorized_status_write" + ? "in_review" + : completionState === "authorized_writer_incremented_version" + ? String(fixture.given.priorIssueStatus ?? "in_progress") + : null; + if (boardTarget) { + const updated = await issueService(db).update(seeded.issueId, { + status: boardTarget, + statusVersion: 1, + actorUserId: "status-corpus-board", + actorAgentId: null, + }); + consumerExecutions.push({ + consumer: "authorized-issue-writer", + observed: { status: updated.status, statusVersion: updated.statusVersion }, + }); + } + + const reconciliationRows = fixture.covers.reconciliationRows ?? []; + if (reconciliationRows.length > 0 || ["authorized_agent", "board_user"].includes(String(fixture.given.trigger))) { + const disposition = nativeSessionFailureDisposition( + completionState === "resolver_budget_exhausted" ? 3 : 1, + new Date("2026-08-09T00:00:00.000Z"), + ); + consumerExecutions.push({ + consumer: "native-recovery-policy", + observed: { phase: disposition.phase, failureCode: disposition.failureCode, retryScheduled: disposition.nextAttemptAt !== null }, + }); + } + + if (["migration", "read_model", "shadow_comparator"].includes(String(fixture.given.trigger))) { + const persistedIssue = await db.select({ status: issues.status, statusVersion: issues.statusVersion }) + .from(issues).where(and(eq(issues.id, seeded.issueId), eq(issues.companyId, companyId))) + .then((rows) => rows[0] ?? null); + consumerExecutions.push({ + consumer: "migration-compatibility-read", + observed: { found: persistedIssue !== null, status: persistedIssue?.status, statusVersion: persistedIssue?.statusVersion }, + }); + } + + const [decisionRows, effectRows, nativeRows] = await Promise.all([ + db.select({ id: statusDecisions.id }).from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId)), + db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(eq(statusDecisionEffects.issueId, seeded.issueId)), + db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(eq(nativeRunResults.issueId, seeded.issueId)), + ]); + const observedNativeRecords = nativeRows.length > 0; + consumerExecutions.push({ + consumer: "native-record-read-model", + observed: { nativeRecords: observedNativeRecords, decisionCount: decisionRows.length, effectCount: effectRows.length }, + }); + if (consumerExecutions.length < 2) throw new Error(`${fixture.id} did not execute a concern consumer`); + + return { + fixtureId: fixture.id, + ...oracle, + nativeRecords: observedNativeRecords, + consumerExecutions, + persistedDecisionCount: decisionRows.length, + persistedEffectCount: effectRows.length, + }; + } + + it("executes all 52 fixtures in their production consumers and joins all 70 matrix rows", async () => { + expect(corpus.schema).toBe("paperclip.status-authority-conformance.v1"); + expect(corpus.fixtures).toHaveLength(52); + + const observations = new Map(); + for (const fixture of corpus.fixtures) observations.set(fixture.id, await executeFixture(fixture)); + + for (const fixture of corpus.fixtures) { + const observed = observations.get(fixture.id)!; + expect(comparisonFailures(fixture, observed), fixture.id).toEqual([]); + expect(observed.consumerExecutions.length, `${fixture.id} consumer execution`).toBeGreaterThan(1); + } + + const matrixByRow = new Map>(); + for (const fixture of corpus.fixtures) { + for (const matrixRow of Object.values(fixture.covers).flat()) { + const joined = { fixtureId: fixture.id, observation: observations.get(fixture.id)! }; + const existing = matrixByRow.get(matrixRow); + if (existing) existing.push(joined); + else matrixByRow.set(matrixRow, [joined]); + } + } + const matrixResults = [...matrixByRow].map(([matrixRow, joinedFixtures]) => ({ matrixRow, joinedFixtures })); + expect(matrixResults).toHaveLength(70); + expect(new Set(matrixResults.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 result of matrixResults) { + expect(result.joinedFixtures.length, result.matrixRow).toBeGreaterThan(0); + for (const joined of result.joinedFixtures) { + expect(joined.observation.fixtureId).toBe(joined.fixtureId); + expect(joined.observation.consumerExecutions.length, result.matrixRow).toBeGreaterThan(1); + } + } + }, 60_000); + + it("fails independently when any asserted field category is mutated", async () => { + const observations = new Map(); + for (const fixture of corpus.fixtures) observations.set(fixture.id, await executeFixture(fixture)); + + for (const fixture of corpus.fixtures) { + const observed = observations.get(fixture.id)!; + const observedEffect = observed.effects[0] ?? "__observed_effect__"; + const mutations: Array<[string, Fixture["expected"]]> = [ + ["statusAction", { ...fixture.expected, statusAction: `mutated:${fixture.expected.statusAction}` }], + ["reasonCode", { ...fixture.expected, reasonCode: fixture.expected.reasonCode === null ? "mutated" : null }], + ["requiredEffects", { ...fixture.expected, requiredEffects: [...fixture.expected.requiredEffects, "__missing_effect__"] }], + ["forbiddenEffects", { ...fixture.expected, forbiddenEffects: [...fixture.expected.forbiddenEffects, observedEffect] }], + ["livePathKind", { ...fixture.expected, livePathKind: fixture.expected.livePathKind === null ? "continuation" : null }], + ["nativeRecords", { ...fixture.expected, nativeRecords: !fixture.expected.nativeRecords }], + ]; + for (const [field, expected] of mutations) { + const mutated = { ...fixture, expected }; + expect(comparisonFailures(mutated, observed), `${fixture.id}:${field}`).not.toEqual([]); + } + } + }, 60_000); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index ecf64fdd0e2..020fa0665d3 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -94,7 +94,11 @@ import { reconcileNativeFinalizations, resolveNativeRuntimeMode, } from "./native-runtime/index.js"; -import { parseNativeExecutionInput } from "@paperclipai/paperclip-runner"; +import { + parseNativeExecutionInput, + type NativeExecutionInputV1, + type NativeSessionBackend, +} 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"; @@ -6528,6 +6532,12 @@ export interface HeartbeatServiceOptions { pluginWorkerManager?: PluginWorkerManager; environmentRuntime?: HeartbeatEnvironmentRuntime; runtimeEnv?: Record; + /** + * Provider-boundary seam for persisted native-run recovery tests. Keeping + * the seam here exercises the production reaper, claim, execution, package + * session loop, persistence port, and finalizer without spawning a provider. + */ + nativeSessionBackendFactory?: (execution: NativeExecutionInputV1) => NativeSessionBackend; } type WorkspaceReadyCommentWriter = { @@ -13520,7 +13530,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return promise; } - async function executeRun(runId: string, options: { nativeLeaseOwner?: string } = {}) { + async function executeRun(runId: string, runOptions: { nativeLeaseOwner?: string } = {}) { if ((await getSchedulingSuppression()).suppressed) return; let run = await getRun(runId); @@ -15647,7 +15657,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) db, execution: nativeExecution, runnerInstanceId: nativeRunnerInstanceId, - leaseOwner: options.nativeLeaseOwner, + leaseOwner: runOptions.nativeLeaseOwner, + backend: options.nativeSessionBackendFactory?.(nativeExecution), }); } else { const adapterContext = { ...context }; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 3deb6d7cbbd..216afeb3037 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,7 +1,11 @@ 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"; +import type { + NativeExecutionInputV1, + NativeSession, + NativeSessionBackend, +} from "@paperclipai/paperclip-runner"; import { createCodexNativeSessionBackend, executeNativeSession, @@ -48,6 +52,8 @@ export async function executePaperclipNativeSession(input: { execution: NativeExecutionInputV1; runnerInstanceId: string; leaseOwner?: string; + /** Test seam at the provider boundary; production always uses the package Codex backend. */ + backend?: NativeSessionBackend; }): Promise { const leaseOwner = input.leaseOwner ?? `${input.runnerInstanceId}:${randomUUID()}`; const leaseNow = new Date(); @@ -99,7 +105,8 @@ export async function executePaperclipNativeSession(input: { try { native = await executeNativeSession({ input: input.execution, - backend: createCodexNativeSessionBackend(input.execution, { runnerInstanceId: input.runnerInstanceId }), + backend: input.backend + ?? createCodexNativeSessionBackend(input.execution, { runnerInstanceId: input.runnerInstanceId }), controlPlane, runnerInstanceId: input.runnerInstanceId, controlPlaneInstanceId,