diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 14807e2186..b1215504a9 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -895,6 +895,16 @@ Native admission verifies local process identities for local runs. Remote runs instead require a provider termination receipt for every lease, with successful cleanup and no active ownership. This applies to both per-turn and warm native runners. A stop receipt retires only the settled cleanup owner for that exact company, run, provider, and sandbox resource, without changing its checkpoint or recorded action outcomes. Independent remote sandboxes have separate cleanup gates, including when one run owns multiple sandboxes. Successful pending-cleanup retries persist the same receipt and reconsider deferred user messages; a delivery failure never reverts successful provider cleanup. A failed checkpoint does not prevent destruction of a terminal run's isolated sandbox; busy ownership still prevents it. +Older local Codex runs may have lost their process IDs before stop receipts were +introduced. A saved post-stop user message may reconcile this legacy case from +the exact retained session: the old controller must be gone, all reconnect +credentials expired, the runner suspended, and the complete startup journal must +identify every provider process in that run. Every recorded process and process +group must be absent, and the retained files must remain unchanged. This path +never launches the old runner or replays its commands. A newer launch record, +incomplete inventory, active process, or unknown state keeps the hold. The queue +keeps the server's wait explanation through normalization and optimistic updates. + Missing receipts and failed cleanup retain the hold. Older providers that return no receipt remain supported but cannot authorize remote continuation. A terminal database status or a PID check on the wrong host is insufficient. diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index a535c0c867..86062b6176 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -3,7 +3,7 @@ import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_ import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; -import { beforeAll, afterAll, describe, it, expect } from "vitest"; +import { beforeAll, afterAll, describe, it, expect, vi } from "vitest"; import { approvals, issueApprovals, issueThreadInteractions, agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, @@ -49,6 +49,33 @@ const support = await getEmbeddedPostgresTestSupport(); agentId: f.agentId, status: "queued", contextSnapshot: { issueId: f.issueId, previousRunId: result.previousRunId, forceFreshSession: true } }); return result; }); + it("reconciles legacy stop evidence transactionally and never uses it for a newer launch", async () => { + const runtime = await import("./native-runtime/native-session-executor.js"); + const verify = vi.spyOn(runtime, "verifyRetainedLocalProcessStop").mockReturnValue({ + fingerprint: "verified-retained-snapshot", providerProcessIds: [999999998], controllerPid: 999999999, + }); + try { + const f = await seed(); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const [environment] = await db.insert(environments).values({ name: "Local legacy", driver: "process" }).returning(); + await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId, + environmentId: environment.id, provider: "local", status: "failed", leasePolicy: "ephemeral", releasedAt: new Date() }); + expect(await admit(f, true)).toMatchObject({ previousRunId: f.sourceRunId }); + expect(await hasNativeLocalProcessStop(db, f.companyId, f.sourceRunId)).toBe(false); + expect(await admit(f)).toMatchObject({ previousRunId: f.sourceRunId }); + expect(await hasNativeLocalProcessStop(db, f.companyId, f.sourceRunId)).toBe(true); + expect(await admit(f)).toBeNull(); + + const next = await seed(); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, next.sourceRunId)); + await appendHeartbeatRunEvent(db, { companyId: next.companyId, runId: next.sourceRunId, + agentId: next.agentId, eventType: PROCESS_START_REQUESTED }); + verify.mockClear(); + expect(await admit(next)).toBeNull(); + expect(verify).not.toHaveBeenCalled(); + } finally { verify.mockRestore(); } + }); + it("preserves local stop proof after process metadata is cleared and invalidates it on another launch", async () => { const f = await seed(); const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 7e99eaa317..8e68e80aad 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,4 +1,4 @@ -import { hasNativeLocalProcessStop } from "./native-local-process-stop.js"; +import { hasNativeLocalProcessStop, reconcileLegacyNativeLocalStop } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; import { z } from "zod"; @@ -103,7 +103,8 @@ export async function admitExplicitNativeContinuation(input: { if (!unusedAdmission) { // A missing process identity is not evidence that a provider exited. if (!run.processPid && !run.processGroupId && - !await hasNativeLocalProcessStop(db, companyId, run.id)) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet."); + !await hasNativeLocalProcessStop(db, companyId, run.id) && + !await reconcileLegacyNativeLocalStop(db, run, coordinator, input.dryRun === true)) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet."); if (run.processPid && !processStopped(run.processPid)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically."); if (run.processGroupId && !processStopped(-run.processGroupId)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically."); } diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index da8b952252..2d0353efbb 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -1,5 +1,5 @@ import { and, desc, eq, inArray, isNull } from "drizzle-orm"; -import { environmentLeases, heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; export const PROCESS_START_REQUESTED = "native.process_start_requested"; @@ -58,3 +58,34 @@ export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId .limit(1); return event?.eventType === LOCAL_PROCESS_STOPPED; } + +/** Upgrade old cleared identities only from an exact, closed retained session. + * New-format launches must use their normal stop receipt, never this fallback. + */ +export async function reconcileLegacyNativeLocalStop( + db: Db, run: typeof heartbeatRuns.$inferSelect, + coordinator: typeof nativeRunFinalizations.$inferSelect | undefined, + dryRun: boolean, +): Promise { + if (!coordinator) return false; + const [newFormat] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + isNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED]), + )).limit(1); + if (newFormat) return false; + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )); + if (!leases.length || leases.some(lease => lease.provider !== "local" || !lease.releasedAt || lease.cleanupStatus === "failed")) return false; + const { verifyRetainedLocalProcessStop } = await import("./native-runtime/native-session-executor.js"); + const proof = verifyRetainedLocalProcessStop(run, coordinator); + if (!proof) return false; + if (!dryRun) await appendHeartbeatRunEvent(db, { + companyId: run.companyId, runId: run.id, agentId: run.agentId, + eventType: LOCAL_PROCESS_STOPPED, stream: "system", level: "info", + message: "Verified the stopped local provider from its retained session before continuing the user message.", + payload: { ...proof, source: "retained_local_session_v1" }, + }); + return true; +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 63ecfa3f5d..f48bc3a8be 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { retainedLocalProviderProcesses } from "./retained-local-stop.js"; import { PROCESS_START_REQUESTED } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { resolveConnectorAssignments, isConnectorSkill } from "../connector-runtime.js"; @@ -1635,6 +1636,55 @@ function cleanupStateSnapshot(root: string) { }; } +/** Read-only compatibility proof for failed local Codex runs whose old recovery + * cleared process metadata. Caller holds the issue/coordinator locks and checks + * local leases and the absence of newer server launch records. No old command + * is executed, no session is resumed, and no unknown action outcome is changed. + */ +export function verifyRetainedLocalProcessStop( + run: typeof heartbeatRuns.$inferSelect, + coordinator: typeof nativeRunFinalizations.$inferSelect, +): { fingerprint: string; providerProcessIds: number[]; controllerPid: number } | null { + try { + if (run.runtimeMode !== "native" || run.status !== "failed" || !run.finishedAt || + run.errorCode !== "native_runner_process_exited" || run.processPid || run.processGroupId || + !run.nativeIssueId || !run.nativeSessionId || !run.runnerInstanceId || + coordinator.companyId !== run.companyId || coordinator.runId !== run.id || + coordinator.issueId !== run.nativeIssueId || coordinator.phase !== "terminal_failure" || + coordinator.leaseOwner || coordinator.resultId || coordinator.nextAttemptAt || + !Number.isSafeInteger(coordinator.controllerPid) || Number(coordinator.controllerPid) <= 0) return null; + // A stopped control-plane process cannot mint another reconnect ticket. + try { process.kill(coordinator.controllerPid!, 0); return null; } + catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") return null; } + const execution = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput); + if (execution.binding.companyId !== run.companyId || execution.binding.agentId !== run.agentId || + execution.binding.issueId !== run.nativeIssueId || execution.binding.runId !== run.id || + nativeSessionKey(execution) !== run.nativeSessionId || execution.provider.kind !== "codex" || + execution.session.driverKind !== "codex_app_server") return null; + const scope = nativeSessionScopeKey(execution); + if (activeNativeSessions.has(run.id) || executingRunnerdSessionScopes.has(scope) || + initializingSessionToolAuthorities.has(scope) || warmNativeSessions.has(scope)) return null; + // Never migrate, search other scopes, or activate a retained directory here. + const root = scopedRunnerdStateRoot(execution); + const snapshot = cleanupStateSnapshot(root); + const identity = record(snapshot.control.identity); + if (identity.runnerInstanceId !== run.runnerInstanceId || + identity.environmentLeaseId !== execution.binding.executionWorkspaceId || + identity.runId !== run.id || identity.normalizedSessionId !== run.nativeSessionId || + typeof identity.turnId !== "string" || typeof identity.itemId !== "string") return null; + const processes = retainedLocalProviderProcesses({ snapshot, now: new Date(), identity: { + runnerInstanceId: run.runnerInstanceId, environmentLeaseId: execution.binding.executionWorkspaceId, + runId: run.id, normalizedSessionId: run.nativeSessionId, + turnId: identity.turnId, itemId: identity.itemId, + } }); + if (!processes || !processes.every(cleanupProcessAbsent)) return null; + if (cleanupStateSnapshot(root).fingerprint !== snapshot.fingerprint) return null; + return { fingerprint: snapshot.fingerprint, providerProcessIds: processes, controllerPid: coordinator.controllerPid! }; + } catch { + return null; + } +} + function cleanupProcessAbsent(pid: unknown): pid is number { if ( process.platform === "win32" || diff --git a/server/src/services/native-runtime/retained-local-stop.test.ts b/server/src/services/native-runtime/retained-local-stop.test.ts new file mode 100644 index 0000000000..7a71f7b05a --- /dev/null +++ b/server/src/services/native-runtime/retained-local-stop.test.ts @@ -0,0 +1,359 @@ +import { describe, expect, it, vi } from "vitest"; +import { retainedLocalProviderProcesses } from "./retained-local-stop.js"; +import { validatePrpEvent } from "../../vendor/paperclip-runner/index.js"; + +export function retainedStopFixture() { + const identity = { + runnerInstanceId: "runner-1", + environmentLeaseId: "workspace-1", + runId: "run-1", + normalizedSessionId: "session-1", + turnId: "turn-1", + itemId: "item-1", + }; + const events = ["intent", "spawned", "intent", "spawned"].map( + (phase, index) => { + const generation = index < 2 ? 2 : 3; + const event = { + schema: "paperclip.prp.event.v1", + schemaVersion: 1, + eventType: "harness.diagnostic", + sourceKind: "runner", + sourceInstanceId: identity.runnerInstanceId, + sourceEventId: `event-${index + 1}`, + sourceSeq: index + 1, + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + turnId: identity.turnId, + itemId: identity.itemId, + priority: 1, + emittedAt: "2026-09-11T10:00:00.000Z", + payload: { + startup: { + schema: "paperclip.provider_startup.v1", + origin: { ...identity }, + launchId: `launch-${generation}`, + attemptedProcessGeneration: generation, + phase, + processId: phase === "spawned" ? 999999990 + generation : null, + processGroupId: phase === "spawned" ? 999999990 + generation : null, + }, + }, + }; + return { + sourceSeq: index + 1, + envelope: { ...identity, payload: event }, + }; + }, + ); + const snapshot = { + fingerprint: "fixture", + fileSha256: ["a", "b", "c"] as const, + control: { + schema: "paperclip.runner.durable.control-plane-state.v1", + identity: { ...identity }, + tickets: { ticket: { expiresAt: "2026-09-11T10:01:00Z" } }, + leases: { lease: { expiresAt: "2026-09-11T11:00:00Z" } }, + commands: [ + { type: "session.snapshot", status: "pending", controllerSeq: 14 }, + ], + committedEvents: events, + ackedSourceSeq: events.length, + }, + runner: { + schema: "paperclip.runner.durable.state.v1", + ...identity, + lifecycle: "suspended", + pendingTerminalDelivery: null as unknown, + pendingProviderCleanup: null as unknown, + outbox: [] as unknown[], + ackedSourceSeq: events.length, + nextSourceSeq: events.length + 1, + lastControllerCommandSeq: 13, + }, + provider: { + schema: "paperclip.runner.codex-provider-state.v1", + lifecycle: "prepared", + startupAttempt: null as unknown, + providerProcessGeneration: 3, + pendingEvents: [] as unknown[], + queuedEvents: [] as unknown[], + toolBridge: { pending: {} }, + activeProviderTurnId: null as string | null, + ambiguousTurnStartPending: false, + }, + }; + return { identity, snapshot, now: new Date("2026-09-11T12:00:00Z") }; +} + +describe("retained local provider inventory", () => { + it("recovers every recorded launch from the complete suspended journal", () => { + const f = retainedStopFixture(); + for (const entry of f.snapshot.control.committedEvents) + expect( + validatePrpEvent(entry.envelope.payload), + JSON.stringify(validatePrpEvent(entry.envelope.payload)), + ).toMatchObject({ ok: true }); + expect(retainedLocalProviderProcesses(f)).toEqual([999999992, 999999993]); + }); + it.each([ + "foreign_scope", + "foreign_event", + "foreign_launch", + "live_runner", + "pending_delivery", + "outbox", + "live_provider", + "startup_intent", + "active_turn", + "ambiguous_turn", + "ticket", + "lease", + "missing_event", + "new_generation", + "missing_spawn", + "bad_pid", + "wrong_group", + "pending_launch", + "bad_schema", + "missing_credentials", + "duplicate_sequence", + "provider_cleanup", + "uncommitted_provider_event", + ])("rejects incomplete or unsettled evidence: %s", (kind) => { + const f = retainedStopFixture(); + const { control, runner, provider } = f.snapshot; + const event = control.committedEvents[0]!.envelope.payload; + const spawned = + control.committedEvents[3]!.envelope.payload.payload.startup; + switch (kind) { + case "uncommitted_provider_event": + provider.pendingEvents.push({ eventType: "harness.diagnostic" }); + break; + case "foreign_scope": + control.identity.runId = "other"; + break; + case "foreign_event": + event.sourceInstanceId = "other"; + break; + case "foreign_launch": + event.payload.startup.origin.runId = "other"; + break; + case "live_runner": + runner.lifecycle = "ready"; + break; + case "pending_delivery": + runner.pendingTerminalDelivery = {}; + break; + case "provider_cleanup": + runner.pendingProviderCleanup = {}; + break; + case "outbox": + runner.outbox.push({}); + break; + case "live_provider": + provider.lifecycle = "turn_active"; + break; + case "startup_intent": + provider.startupAttempt = {}; + break; + case "active_turn": + provider.activeProviderTurnId = "turn"; + break; + case "ambiguous_turn": + provider.ambiguousTurnStartPending = true; + break; + case "ticket": + control.tickets.ticket.expiresAt = "2026-09-12T00:00:00Z"; + break; + case "lease": + control.leases.lease.expiresAt = "invalid"; + break; + case "missing_credentials": + control.leases = {} as typeof control.leases; + break; + case "missing_event": + control.committedEvents.shift(); + break; + case "new_generation": + provider.providerProcessGeneration++; + break; + case "missing_spawn": + spawned.phase = "intent"; + break; + case "bad_pid": + spawned.processId = -1; + break; + case "wrong_group": + spawned.processGroupId = 42; + break; + case "pending_launch": + control.commands[0]!.type = "run.attach"; + break; + case "bad_schema": + runner.schema = "unknown"; + break; + case "duplicate_sequence": + event.sourceSeq = 2; + break; + } + expect(retainedLocalProviderProcesses(f)).toBeNull(); + }); +}); + +describe("retained local stop host verification", () => { + it.each([ + "stopped", + "live_controller", + "live_provider", + "permission_denied", + "foreign_company", + "newer_lease", + "symlink", + "missing_file", + ])( + "checks persisted scope and real host evidence: %s", + async (kind) => { + const { mkdtempSync, mkdirSync, writeFileSync, rmSync, symlinkSync } = + await import("node:fs"); + const { tmpdir } = await import("node:os"); + const { join } = await import("node:path"); + const { createHash } = await import("node:crypto"); + const { canonicalNativeJson } = await import("./canonical.js"); + const { verifyRetainedLocalProcessStop } = + await import("./native-session-executor.js"); + const f = retainedStopFixture(); + const base = mkdtempSync(join(tmpdir(), "retained-stop-")); + const previousBase = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = base; + const execution = { + schema: "paperclip.native-execution-input.v1", + provider: { kind: "codex", model: null }, + binding: { + companyId: "company", + agentId: "agent", + issueId: "issue", + runId: "run-1", + executionWorkspaceId: "workspace-1", + }, + task: { + identifier: "TEST-1", + title: "Continue", + description: null, + prompt: "Continue", + workMode: "standard", + }, + workspace: { + cwd: base, + repoUrl: null, + repoRef: null, + branchName: null, + }, + session: { + normalizedSessionId: "session-1", + driverKind: "codex_app_server", + protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null }, + }, + completionContract: { + id: "contract", + sha256: "sha", + schemaVersion: "paperclip.completion-contract.v1", + contract: { + revision: "1", + objective: "Continue", + criteria: [{ id: "objective", requirement: "Finish" }], + }, + }, + interactionResponses: [], + credentialBindings: [], + }; + const run = { + id: "run-1", + companyId: "company", + agentId: "agent", + nativeIssueId: "issue", + runtimeMode: "native", + status: "failed", + finishedAt: new Date("2026-09-11T11:00:00Z"), + errorCode: "native_runner_process_exited", + processPid: null, + processGroupId: null, + nativeSessionId: "session-1", + runnerInstanceId: "runner-1", + runnerProfileJson: { nativeExecutionInput: execution }, + } as unknown as Parameters[0]; + const coordinator = { + companyId: "company", + issueId: "issue", + runId: "run-1", + phase: "terminal_failure", + controllerPid: kind === "live_controller" ? process.pid : 999999999, + } as Parameters[1]; + const scope = canonicalNativeJson({ + schema: "paperclip.native-session-scope.v2", + companyId: "company", + agentId: "agent", + workspace: { kind: "managed", executionWorkspaceId: "workspace-1" }, + provider: { + driverKind: "codex_app_server", + identity: { kind: "codex" }, + }, + normalizedSessionId: "session-1", + }); + const root = join(base, createHash("sha256").update(scope).digest("hex")); + mkdirSync(join(root, "runner"), { recursive: true }); + mkdirSync(join(root, "control-plane")); + if (kind === "live_provider") { + const startup = + f.snapshot.control.committedEvents[3]!.envelope.payload.payload + .startup; + startup.processId = process.pid; + startup.processGroupId = process.pid; + } + if (kind === "foreign_company") execution.binding.companyId = "other"; + if (kind === "newer_lease") coordinator.leaseOwner = "active"; + for (const [name, value] of [ + ["control-plane/control-plane-state.json", f.snapshot.control], + ["runner/runner-state.json", f.snapshot.runner], + ["runner/codex-provider-state.json", f.snapshot.provider], + ] as const) + writeFileSync(join(root, name), JSON.stringify(value)); + if (kind === "symlink" || kind === "missing_file") { + rmSync(join(root, "runner/codex-provider-state.json")); + if (kind === "symlink") { + writeFileSync( + join(base, "provider.json"), + JSON.stringify(f.snapshot.provider), + ); + symlinkSync( + join(base, "provider.json"), + join(root, "runner/codex-provider-state.json"), + ); + } + } + const kill = + kind === "permission_denied" + ? vi.spyOn(process, "kill").mockImplementation(() => { + throw Object.assign(new Error("denied"), { code: "EPERM" }); + }) + : null; + try { + const result = verifyRetainedLocalProcessStop(run, coordinator); + if (kind === "stopped") + expect(result).toMatchObject({ + providerProcessIds: [999999992, 999999993], + controllerPid: 999999999, + }); + else expect(result).toBeNull(); + } finally { + kill?.mockRestore(); + if (previousBase === undefined) + delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previousBase; + rmSync(base, { recursive: true, force: true }); + } + }, + 30000, + ); +}); diff --git a/server/src/services/native-runtime/retained-local-stop.ts b/server/src/services/native-runtime/retained-local-stop.ts new file mode 100644 index 0000000000..4615a46e7d --- /dev/null +++ b/server/src/services/native-runtime/retained-local-stop.ts @@ -0,0 +1,187 @@ +import { validatePrpEvent } from "../../vendor/paperclip-runner/index.js"; +import type { + RetainedMaintenanceIdentity, + RetainedMaintenanceSnapshot, +} from "./native-maintenance-no-launch.js"; + +const record = (value: unknown): Record => { + if (!value || typeof value !== "object" || Array.isArray(value)) + throw new Error("invalid"); + return value as Record; +}; +const array = (value: unknown): unknown[] => { + if (!Array.isArray(value) || value.length > 100_000) + throw new Error("invalid"); + return value; +}; +function requireProof(value: unknown): asserts value { + if (!value) throw new Error("unproven"); +} +const identityKeys = [ + "runnerInstanceId", + "environmentLeaseId", + "runId", + "normalizedSessionId", + "turnId", + "itemId", +] as const; + +/** Inventory only. The caller must separately verify local host ownership, + * terminal coordinator/runner exit, absent processes, and unchanged files. + * Missing launch records never count as evidence that a process stopped. + */ +export function retainedLocalProviderProcesses(input: { + snapshot: RetainedMaintenanceSnapshot; + identity: RetainedMaintenanceIdentity; + now: Date; +}): number[] | null { + try { + const { control, runner, provider } = input.snapshot; + const { identity } = input; + const now = input.now.getTime(); + requireProof(Number.isFinite(now)); + requireProof( + control.schema === "paperclip.runner.durable.control-plane-state.v1", + ); + requireProof(runner.schema === "paperclip.runner.durable.state.v1"); + requireProof( + provider.schema === "paperclip.runner.codex-provider-state.v1", + ); + for (const key of identityKeys) { + requireProof( + typeof identity[key] === "string" && identity[key].length > 0, + ); + requireProof( + record(control.identity)[key] === identity[key] && + runner[key] === identity[key], + ); + } + requireProof( + runner.lifecycle === "suspended" && + runner.pendingTerminalDelivery === null, + ); + requireProof( + runner.pendingProviderCleanup == null && + array(runner.outbox).length === 0, + ); + requireProof( + provider.lifecycle === "prepared" && provider.startupAttempt === null, + ); + requireProof( + array(provider.pendingEvents).length === 0 && + array(provider.queuedEvents).length === 0, + ); + requireProof( + Object.keys(record(record(provider.toolBridge).pending)).length === 0, + ); + requireProof( + provider.activeProviderTurnId === null && + provider.ambiguousTurnStartPending === false, + ); + // An old controller or a still-valid reconnect credential must not revive + // this authority after the process inventory has been checked. + for (const name of ["tickets", "leases"]) { + const credentials = Object.values(record(control[name])); + requireProof(credentials.length > 0 && credentials.length <= 4096); + for (const raw of credentials) { + const expiry = Date.parse(String(record(raw).expiresAt)); + requireProof(Number.isFinite(expiry) && expiry <= now); + } + } + for (const raw of array(control.commands)) { + const command = record(raw); + requireProof( + ["completed", "failed", "pending"].includes(String(command.status)), + ); + if (command.status === "pending") { + requireProof( + ["session.snapshot", "runner.suspend"].includes(String(command.type)), + ); + requireProof( + Number(command.controllerSeq) > + Number(runner.lastControllerCommandSeq), + ); + } + } + const events = array(control.committedEvents); + requireProof(events.length > 0 && control.ackedSourceSeq === events.length); + requireProof( + runner.ackedSourceSeq === events.length && + runner.nextSourceSeq === events.length + 1, + ); + const launches = new Map< + string, + { generation: number; pid: number | null } + >(); + let lastGeneration = 0; + for (const [index, raw] of events.entries()) { + const wrapper = record(raw); + const envelope = record(wrapper.envelope); + const event = record(envelope.payload); + requireProof(validatePrpEvent(event).ok); + for (const key of identityKeys) + requireProof(envelope[key] === identity[key]); + requireProof( + event.runId === identity.runId && + event.normalizedSessionId === identity.normalizedSessionId, + ); + requireProof( + event.sourceKind === "runner" && + event.sourceInstanceId === identity.runnerInstanceId, + ); + requireProof( + event.turnId === identity.turnId && event.itemId === identity.itemId, + ); + requireProof( + event.sourceSeq === index + 1 && wrapper.sourceSeq === event.sourceSeq, + ); + const startupValue = record(event.payload).startup; + if (startupValue === undefined) continue; + const startup = record(startupValue); + requireProof( + event.eventType === "harness.diagnostic" && + startup.schema === "paperclip.provider_startup.v1", + ); + for (const key of identityKeys.filter( + (key) => key !== "environmentLeaseId", + )) { + requireProof(record(startup.origin)[key] === identity[key]); + } + const launchId = startup.launchId; + const generation = startup.attemptedProcessGeneration; + requireProof(typeof launchId === "string" && launchId.length > 0); + requireProof(Number.isSafeInteger(generation) && Number(generation) > 0); + if (startup.phase === "intent") { + requireProof( + !launches.has(launchId) && Number(generation) > lastGeneration, + ); + requireProof( + startup.processId === null && startup.processGroupId === null, + ); + launches.set(launchId, { generation: Number(generation), pid: null }); + lastGeneration = Number(generation); + } else { + requireProof(startup.phase === "spawned"); + const launch = launches.get(launchId); + requireProof( + launch && launch.generation === generation && launch.pid === null, + ); + requireProof( + Number.isSafeInteger(startup.processId) && + Number(startup.processId) > 0, + ); + requireProof(startup.processGroupId === startup.processId); + launch.pid = Number(startup.processId); + } + } + requireProof( + launches.size > 0 && + provider.providerProcessGeneration === lastGeneration, + ); + const processes = [...launches.values()].map((launch) => launch.pid); + requireProof(processes.every((pid): pid is number => pid !== null)); + return [...new Set(processes)]; + } catch { + return null; + } +} diff --git a/ui/src/lib/issue-queued-comment-queue.test.ts b/ui/src/lib/issue-queued-comment-queue.test.ts index dae5ee5982..204b41af37 100644 --- a/ui/src/lib/issue-queued-comment-queue.test.ts +++ b/ui/src/lib/issue-queued-comment-queue.test.ts @@ -22,6 +22,29 @@ function comment(id: string, body: string) { } describe("normalizeIssueQueuedCommentQueue", () => { + it("keeps the recovery explanation through normalization and optimistic merging", () => { + const wait = { reason: "process_identity_missing", message: "The previous run has no verified stop record." }; + const authoritativeQueue = normalizeIssueQueuedCommentQueue({ + state: "deferred", executionWait: wait, + entries: [{ comment: comment("saved", "continue"), position: 0 }], + }, "issue-1"); + const queue = mergePendingIssueQueuedComments({ issueId: "issue-1", authoritativeQueue, + pendingComments: [{ comment: comment("new", "continue"), targetRunId: null }], + fallbackProtocol: "paperclip_runner_v1" }); + expect(queue?.executionWait).toEqual(wait); + expect(queue?.entries).toHaveLength(2); + const running = normalizeIssueQueuedCommentQueue({ ...authoritativeQueue, state: "queued" }, "issue-1"); + expect(running.executionWait).toBeNull(); + expect(mergePendingIssueQueuedComments({ issueId: "issue-1", authoritativeQueue: running, + pendingComments: [], fallbackProtocol: "paperclip_runner_v1" })?.executionWait).toBeNull(); + }); + + it.each([null, "wait", {}, { reason: 1, message: "Wait" }, { reason: "wait", message: " " }])( + "drops malformed execution wait data: %j", executionWait => { + expect(normalizeIssueQueuedCommentQueue({ state: "deferred", executionWait }, "issue-1").executionWait).toBeNull(); + }, + ); + it("sorts, deduplicates, and drops malformed queue entries", () => { const queue = normalizeIssueQueuedCommentQueue( { diff --git a/ui/src/lib/issue-queued-comment-queue.ts b/ui/src/lib/issue-queued-comment-queue.ts index 52a03547e0..4e4c7e0639 100644 --- a/ui/src/lib/issue-queued-comment-queue.ts +++ b/ui/src/lib/issue-queued-comment-queue.ts @@ -54,6 +54,7 @@ export function normalizeIssueQueuedCommentQueue( .map((entry, position) => ({ ...entry, position })); const disposition = source?.steeringDisposition; const state = source?.state; + const wait = record(source?.executionWait); return { issueId: @@ -79,6 +80,10 @@ export function normalizeIssueQueuedCommentQueue( ) ? (disposition as IssueQueuedCommentSteeringDisposition) : "unsupported", + executionWait: + state === "deferred" && typeof wait?.reason === "string" && + typeof wait.message === "string" && wait.reason.trim() && wait.message.trim() + ? { reason: wait.reason, message: wait.message } : null, entries, }; } @@ -144,6 +149,8 @@ export function mergePendingIssueQueuedComments(params: { (protocol === "paperclip_runner_v1" && targetRunId ? "temporarily_unavailable" : "unsupported"), + executionWait: params.authoritativeQueue?.state === "deferred" + ? params.authoritativeQueue.executionWait ?? null : null, entries, }; }