diff --git a/doc/observability.md b/doc/observability.md index d8c4b0e4a26..e877610b3d0 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -443,6 +443,30 @@ values become `unknown`. A signal such as `SIGKILL` does not establish who sent it or prove an out-of-memory kill. These fields do not change error grouping or run outcomes, and do not include process output or adapter result payloads. +For a native Codex turn that ends with a structured HTTP 400 model/ChatGPT +compatibility rejection, the run records `native_provider_model_rejected` +and a fixed message that asks the user to select a supported model or connection. +This requires a committed runner terminal for the selected model and the exact +failed turn. Agent text, tool output, unrelated turns, and unknown provider +errors do not create this classification. The provider diagnostic contains only +`provider=codex`, `category=model_auth_incompatible`, `status=400`, and +`authMode=chatgpt`. It excludes the model name and provider response. This does +not change credentials, select a fallback model, replay a turn, or alter the +accepted semantic result or cleanup. The existing configuration-blocker recovery +path requires a configuration change instead of automatically retrying the same +model and connection. Native finalization also derives this evidence from the +committed event and the run's pinned provider configuration before it can queue +a failed-result retry. This survives a controller restart before the adapter +diagnostic is saved. Invalid bindings, missing evidence, or a mismatched event +hash do not grant this classification. A current worker's task is blocked for a +configuration change. A pending review instead keeps its original decision and +status version, releases execution, and creates no automatic recovery action. +After repairing the configuration, explicitly retry the reviewer to resolve the +same pending review. The Board's exact failed-run Retry accepts this classified +native failure only while its saved review is still current and pending; caller +payloads cannot select a different review. Dispatch checks that assignment again. +Superseded or completed reviews retain their authority. + An unconfirmed adapter Stop timeout has an event-local `adapter_stop` context: the run UUID, built-in adapter type, native/legacy runtime mode, configured wait duration, and whether abort was requested. Invalid identities become diff --git a/server/src/__tests__/agent-live-run-routes.test.ts b/server/src/__tests__/agent-live-run-routes.test.ts index 55d978e0798..d5d1805a694 100644 --- a/server/src/__tests__/agent-live-run-routes.test.ts +++ b/server/src/__tests__/agent-live-run-routes.test.ts @@ -23,6 +23,7 @@ const mockIssueService = vi.hoisted(() => ({ getById: vi.fn(), getByIdentifier: vi.fn(), })); +const mockNativeReviewAssignment = vi.hoisted(() => vi.fn()); const mockExecutionProjection = vi.hoisted(() => ({ executionProjectionForRun: vi.fn(async () => null), @@ -79,6 +80,10 @@ const mockChatRunRetries = vi.hoisted(() => ({ })); function registerModuleMocks() { + vi.doMock("../services/native-runtime/native-review-participant.js", async () => ({ + ...(await vi.importActual("../services/native-runtime/native-review-participant.js")), + getNativeReviewAssignment: mockNativeReviewAssignment, + })); vi.doMock("../services/execution-projection.js", () => mockExecutionProjection); vi.doMock("../routes/authz.js", async () => vi.importActual("../routes/authz.js"), @@ -287,6 +292,7 @@ describe("agent live run routes", () => { vi.clearAllMocks(); mockChatRunRetries.prepareFailedChatRunRetry.mockReset(); mockChatRunRetries.processFailedChatRunRetry.mockReset(); + mockNativeReviewAssignment.mockReset().mockResolvedValue(null); mockAccessService.canUser.mockResolvedValue(true); mockAccessService.decide.mockImplementation(async (input: { action?: string }) => ({ allowed: true, @@ -966,6 +972,68 @@ describe("agent live run routes", () => { expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(routeAgentId, expect.objectContaining({ manualUserWake: true })); }); + describe("exact native model-rejected review retry", () => { + const reviewContext = { + nativeReviewInteractionId: "55555555-5555-4555-8555-555555555555", + nativeReviewDecisionId: "66666666-6666-4666-8666-666666666666", + }; + const selectedRun = { + id: failedChatRunId, companyId: "company-1", agentId: routeAgentId, + status: "failed", runtimeMode: "native", errorCode: "native_provider_model_rejected", + nativeIssueId: failedChatIssueId, + contextSnapshot: { issueId: failedChatIssueId, ...reviewContext }, + }; + const selectedIssue = { id: failedChatIssueId, companyId: "company-1", status: "in_review", assigneeAgentId: "worker", assigneeUserId: null }; + const retryBody = { failedRunId: failedChatRunId, reason: "retry_failed_run" }; + beforeEach(() => { + mockAgentService.getById.mockResolvedValue({ id: routeAgentId, companyId: "company-1" }); + mockHeartbeatService.getRun.mockResolvedValue(selectedRun); + mockIssueService.getById.mockResolvedValue(selectedIssue); + mockNativeReviewAssignment.mockResolvedValue({ interaction: { status: "pending" } }); + }); + + it("permits the Board to retry the exact pending review using only saved context", async () => { + const fixture = createFailedChatRetryDb(false); + const res = await requestApp(await createApp(fixture.db), url => request(url).post(`/api/agents/${routeAgentId}/wakeup`).send({ + ...retryBody, payload: { issueId: "forged", nativeReviewInteractionId: "forged", nativeReviewDecisionId: "forged" }, + })); + expect(res.status, JSON.stringify(res.body)).toBe(202); + expect(mockNativeReviewAssignment).toHaveBeenCalledExactlyOnceWith(fixture.db, { + companyId: "company-1", issueId: failedChatIssueId, agentId: routeAgentId, contextSnapshot: reviewContext, + }); + expect(mockHeartbeatService.wakeup).toHaveBeenCalledExactlyOnceWith(routeAgentId, expect.objectContaining({ + failedRunId: failedChatRunId, reason: "retry_failed_run", payload: { issueId: failedChatIssueId }, + requestedByActorType: "user", contextSnapshot: expect.objectContaining(reviewContext), + })); + expect(JSON.stringify(mockHeartbeatService.wakeup.mock.calls)).not.toContain("forged"); + }); + + it.each(["agent-caller", "denied-task", "foreign-company", "foreign-agent", "foreign-issue", "legacy", "unknown-error", "missing-context", "stale", "resolved", "changed-reviewer"])( + "rejects an ineligible exact review retry (%s)", async scenario => { + let actor: Record | undefined; + if (scenario === "agent-caller") actor = { type: "agent", agentId: routeAgentId, companyId: "company-1" }; + if (scenario === "denied-task") mockAccessService.decide.mockImplementation(async ({ action }) => ({ allowed: action !== "issue:comment", explanation: "Denied" })); + if (scenario === "foreign-company") mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, companyId: "company-2" }); + if (scenario === "foreign-agent") mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, agentId: "other-reviewer" }); + if (scenario === "foreign-issue") mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, nativeIssueId: "other-task" }); + if (scenario === "legacy") mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, runtimeMode: "legacy" }); + if (scenario === "unknown-error") mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, errorCode: "adapter_failed" }); + if (scenario === "missing-context") mockHeartbeatService.getRun.mockResolvedValue({ ...selectedRun, contextSnapshot: { issueId: failedChatIssueId } }); + if (scenario === "stale" || scenario === "changed-reviewer") mockNativeReviewAssignment.mockResolvedValue(null); + if (scenario === "resolved") mockNativeReviewAssignment.mockResolvedValue({ interaction: { status: "accepted" } }); + const fixture = createFailedChatRetryDb(false); + const res = await requestApp(await createApp(fixture.db, actor), url => request(url).post(`/api/agents/${routeAgentId}/wakeup`).send({ + ...retryBody, payload: { ...reviewContext }, + })); + expect(res.status).toBe(["agent-caller", "denied-task"].includes(scenario) ? 403 : ["foreign-company", "foreign-agent"].includes(scenario) ? 404 : 409); + expect(mockHeartbeatService.wakeup).not.toHaveBeenCalled(); + if (["agent-caller", "denied-task", "foreign-company", "foreign-agent", "foreign-issue", "legacy", "unknown-error", "missing-context"].includes(scenario)) { + expect(mockNativeReviewAssignment).not.toHaveBeenCalled(); + } + }, + ); + }); + describe("exact failed chat run retry", () => { const retryBody = { failedRunId: failedChatRunId, diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 80d2011761f..88c1a64cfeb 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -14310,7 +14310,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { await commitNativeStatusDecision({ db, companyId: fixture.companyId, issueId: fixture.issueId, runId: fixture.runId, assessmentId: coordinator!.assessmentId!, priorStatus: issue!.status, priorStatusVersion: issue!.statusVersion, priorDecisionId: issue!.lastStatusDecisionId, - decision: { policyVersion: "phase6-v7", statusAction: "in_progress", toStatus: "in_progress", + decision: { policyVersion: "phase6-v8", statusAction: "in_progress", toStatus: "in_progress", reasonCode: "board_response_waiting", unblockDescriptor: null, effects: [] }, requireBoardResponseWaitSource: source!.source, requireBoardResponseWaitOrigin: origin!, }); @@ -14934,7 +14934,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { priorStatusVersion: issue!.statusVersion, priorDecisionId: issue!.lastStatusDecisionId, decision: { - policyVersion: "phase6-v7", + policyVersion: "phase6-v8", statusAction: reasonCode === "prior_status_preserved_no_live_path" ? "preserve" : "in_progress", toStatus: "in_progress", reasonCode, diff --git a/server/src/modules/wake-queue/domain/values.ts b/server/src/modules/wake-queue/domain/values.ts index 48191b69d2a..fc4fa8fa5e2 100644 --- a/server/src/modules/wake-queue/domain/values.ts +++ b/server/src/modules/wake-queue/domain/values.ts @@ -22,5 +22,6 @@ export function isWorkspaceValidationFailedRun(run: { errorCode: string | null } } export function isConfigurationIncompleteFailedRun(run: { errorCode: string | null }): boolean { - return run.errorCode === CONFIGURATION_INCOMPLETE_FAILURE_CODE || run.errorCode === "model_not_found"; + return run.errorCode === CONFIGURATION_INCOMPLETE_FAILURE_CODE || run.errorCode === "model_not_found" || + run.errorCode === "native_provider_model_rejected"; } diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index 93ab497d325..6e4ce96fd2b 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -15,6 +15,7 @@ import { isAiConnectionCompatible } from "@paperclipai/shared"; import { applyConnectorSkills, resolveConnectorAssignments, annotateConnectorSkills, isConnectorSkill } from "../services/connector-runtime.js"; import { getExecutionBlocker } from "../services/execution-blocker.js"; import { canRetryStoppedRun } from "../services/cancelled-native-startup.js"; +import { getNativeReviewAssignment, readNativeReviewAssignmentContext, type NativeReviewAssignmentContext } from "../services/native-runtime/native-review-participant.js"; import { paperclipRunnerTransitionConfig, normalizeLegacyRunnerProvider, isPaperclipRunnerProvider } from "@paperclipai/adapter-utils"; import { executionProjectionForRun, executionProjectionsForRuns } from "../services/execution-projection.js"; import { selectDashboardRunIds } from "../services/dashboard-run-selection.js"; @@ -6036,6 +6037,7 @@ export function agentRoutes( let wakePayload = req.body.payload ?? null; let retryConversationContext: Record = {}; + let retryNativeReviewContext: NativeReviewAssignmentContext | null = null; if (req.body.failedRunId) { assertBoard(req); if ( @@ -6089,7 +6091,22 @@ export function agentRoutes( }, }); if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); - if (issue.assigneeAgentId !== agent.id) throw conflict("The task is no longer assigned to this agent."); + const savedReviewContext = readNativeReviewAssignmentContext(failedContext); + if (failedRun.runtimeMode === "native" && failedRun.status === "failed" && + failedRun.errorCode === "native_provider_model_rejected" && + failedRun.nativeIssueId === issue.id && savedReviewContext) { + const review = await getNativeReviewAssignment(db, { + companyId: agent.companyId, issueId: issue.id, agentId: agent.id, + contextSnapshot: savedReviewContext, + }); + if (review?.interaction.status !== "pending") { + throw conflict("The failed run's review is no longer current."); + } + retryNativeReviewContext = savedReviewContext; + } + if (issue.assigneeAgentId !== agent.id && !retryNativeReviewContext) { + throw conflict("The task is no longer assigned to this agent."); + } if (issue.conversationAgentId) { // Agent Chat has no task description to replay. Recover the exact // request from the selected server-owned run, never caller markers. @@ -6196,6 +6213,9 @@ export function agentRoutes( requestedByActorId: req.actor.type === "agent" ? req.actor.agentId ?? null : req.actor.userId ?? null, contextSnapshot: { ...retryConversationContext, + // Only the selected server-owned run can carry review authority into + // an exact retry. Dispatch revalidates the current review assignment. + ...retryNativeReviewContext, triggeredBy: req.actor.type, originIdentityContextId: req.actor.identityContextId ?? null, responsibleUserId: req.actor.type === "agent" ? req.actor.onBehalfOfUserId ?? null : req.actor.userId ?? null, diff --git a/server/src/services/__tests__/run-failure-diagnostics.test.ts b/server/src/services/__tests__/run-failure-diagnostics.test.ts index 98a585aa7c4..6ee721b7a37 100644 --- a/server/src/services/__tests__/run-failure-diagnostics.test.ts +++ b/server/src/services/__tests__/run-failure-diagnostics.test.ts @@ -7,6 +7,19 @@ const run = (overrides: Partial = {}) => ({ resultJson: null, ...overrides const collect = (error: unknown) => collectRunFailureDiagnostics(run(), { error }); describe("run failure diagnostics", () => { + it("exports only the closed native model/auth rejection vocabulary", () => { + const diagnostic = { provider: "codex", category: "model_auth_incompatible", status: 400, authMode: "chatgpt" }; + const result = sanitizeRunFailureDiagnostics(collectRunFailureDiagnostics(run({ resultJson: { + nativeProviderFailure: { ...diagnostic, model: "private-model", response: "private-provider-response", prompt: "private-prompt" }, + } }), {})); + expect(result.provider).toEqual(diagnostic); + expect(JSON.stringify(result)).not.toContain("private"); + for (const value of [null, [], { ...diagnostic, category: "private-category" }, + Object.defineProperty({}, "provider", { get() { throw new Error("private"); } })]) { + expect(collectRunFailureDiagnostics(run({ resultJson: { nativeProviderFailure: value } }), {}).provider).toEqual({}); + } + }); + it("selects bounded lock-owner evidence from a caught timeout cause", () => { const error = new Error("outer", { cause: Object.assign(new Error("lock timeout"), { code: "ERR_WORKSPACE_RESTORE_LOCK_TIMEOUT", diff --git a/server/src/services/native-runtime/native-provider-failure-evidence.ts b/server/src/services/native-runtime/native-provider-failure-evidence.ts new file mode 100644 index 00000000000..e97143aca9e --- /dev/null +++ b/server/src/services/native-runtime/native-provider-failure-evidence.ts @@ -0,0 +1,45 @@ +import { and, desc, eq, sql } from "drizzle-orm"; +import { heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; +import { parseNativeExecutionInput, validatePrpEvent, type PrpTerminalState } from "../../vendor/paperclip-runner/index.js"; +import { nativeSha256 } from "./canonical.js"; +import { createNativeProviderFailureObservation } from "./native-provider-failure.js"; + +function record(value: unknown): Record { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? value as Record : {}; +} + +/** Re-derive authority from immutable run-bound evidence, including after restart. */ +export async function readPersistedNativeModelRejection( + db: Db, + run: typeof heartbeatRuns.$inferSelect, + turnId: string | null, + terminal: PrpTerminalState, +) { + if (run.runtimeMode !== "native" || !turnId || !run.nativeSessionId || !run.runnerInstanceId || + terminal.turnTerminalState !== "failed" || terminal.runTerminalState !== "failed") return null; + let execution; + try { execution = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput); } catch { return null; } + if (execution.provider.kind !== "codex" || !execution.provider.model || + execution.binding.runId !== run.id || execution.binding.companyId !== run.companyId || + execution.binding.agentId !== run.agentId || execution.binding.issueId !== run.nativeIssueId || + execution.session.normalizedSessionId !== run.nativeSessionId || + execution.completionContract.id !== run.completionContractId || + execution.completionContract.sha256 !== run.completionContractSha256) return null; + + const [row] = await db.select().from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.sourceInstanceId, run.runnerInstanceId), eq(heartbeatRunEvents.eventType, "turn.failed"), + sql`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId' = ${turnId}`, + )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + if (!row) return null; + const parsed = validatePrpEvent(record(row.payload).prpEvent); + if (!parsed.ok || parsed.event.runId !== run.id || parsed.event.normalizedSessionId !== run.nativeSessionId || + parsed.event.sourceKind !== "runner" || parsed.event.sourceInstanceId !== run.runnerInstanceId || + parsed.event.sourceEventId !== row.sourceEventId || parsed.event.sourceSeq !== row.sourceSeq || + parsed.event.eventType !== row.eventType || parsed.event.turnId !== turnId || + row.sourcePayloadSha256 !== nativeSha256(parsed.event)) return null; + const observed = createNativeProviderFailureObservation(execution.provider); + observed.observe(parsed.event); + return observed.forTerminal(turnId, terminal); +} diff --git a/server/src/services/native-runtime/native-provider-failure.integration.test.ts b/server/src/services/native-runtime/native-provider-failure.integration.test.ts new file mode 100644 index 00000000000..2da7891abbc --- /dev/null +++ b/server/src/services/native-runtime/native-provider-failure.integration.test.ts @@ -0,0 +1,272 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issueThreadInteractions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, workAssessments } from "@paperclipai/db"; +import type { ControlPlanePort, NativeExecutionInputV1, PrpEvent, PrpTerminalState } from "@paperclipai/paperclip-runner"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT } from "../../vendor/paperclip-runner/testing.js"; + +const provider = vi.hoisted(() => ({ execute: vi.fn() })); +vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({ + ...(await importOriginal()), + executeNativeSession: provider.execute, +})); + +import { executePaperclipNativeSession } from "./native-session-executor.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { NATIVE_MODEL_REJECTION_DIAGNOSTIC, NATIVE_MODEL_REJECTION_MESSAGE } from "./native-provider-failure.js"; +import { collectRunFailureDiagnostics } from "../run-failure-diagnostics.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { readPersistedNativeModelRejection } from "./native-provider-failure-evidence.js"; +import { heartbeatService } from "../heartbeat.js"; +import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; +import { claimQueuedNativeReviewRun } from "./native-review-dispatch.js"; +import { commitNativeStatusDecision, NativeStatusRaceError } from "./status-decision-committer.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "./status-arbiter.js"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; +import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; +import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; + +describe("native completed failure diagnostics", () => { + let temporary: Awaited>; + let db: ReturnType; + const companyId = randomUUID(); + const agentId = randomUUID(); + const workerId = randomUUID(); + const model = "test-model"; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-model-failure-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Native failure regression", issuePrefix: "NMF" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native failure agent", status: "active", adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model } }); + await db.insert(agents).values({ id: workerId, companyId, name: "Review source worker", status: "active", adapterType: "paperclip_runner" }); + }); + afterAll(async () => { await temporary?.cleanup(); }); + + it.each(["committed", "duplicate", "pending-review", "superseded-review", "resolved-review", "reassigned", "successor-race", "different-turn", "unknown-error", "untrusted-source", "untrusted-run", "untrusted-session", "succeeded", "cancelled"] as const)( + "preserves accepted results and only labels the matching provider rejection (%s)", async (scenario) => { + const issueId = randomUUID(); + const runId = randomUUID(); + const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Explain a provider rejection", status: "in_progress", workMode: "standard", assigneeAgentId: agentId }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: issueId, invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId } }).returning(); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const native = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId)); + const completionInput = buildNativeCompletionContract(issue, { revision: contract.revision }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed" }); + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { companyId, runId, issueId, agentId, executionWorkspaceId: runId }, + provider: { kind: "codex", model }, + task: { identifier: issueId, title: issue.title, description: null, prompt: "Reply to the current message.", workMode: "standard" }, + workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", contract: completionInput }, + interactionResponses: [], credentialBindings: [], + }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, runId)); + if (scenario.endsWith("-review")) { + const sourceRunId = randomUUID(), sourceResultId = randomUUID(), assessmentId = randomUUID(); + const decisionId = randomUUID(), interactionId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId: workerId, nativeIssueId: issueId, runtimeMode: "native", status: "succeeded", completionContractId: contract.id, completionContractSha256: contract.canonicalSha256 }); + await db.insert(nativeRunResults).values({ id: sourceResultId, companyId, issueId, runId: sourceRunId, completionContractId: contract.id, serverFingerprint: sourceResultId, schemaStatus: "accepted", resultJson: {}, canonicalSha256: sourceResultId }); + await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId, runId: sourceRunId, contractId: contract.id, resultId: sourceResultId, triggerKind: "native_result", triggerActorCompanyId: companyId, priorIssueStatus: "in_progress", priorStatusVersion: 0, policyVersion: "fixture", assessmentJson: {}, inputDigest: assessmentId }); + const [reviewIssue] = await db.update(issues).set({ status: "in_review", assigneeAgentId: workerId }).where(eq(issues.id, issueId)).returning(); + await db.insert(statusDecisions).values({ id: decisionId, companyId, issueId, runId: sourceRunId, assessmentId, decisionVersion: 1, policyVersion: "fixture", fromStatus: "in_progress", toStatus: "in_review", reasonCode: "explicit_review", decisionJson: { projectedStatusVersion: reviewIssue.statusVersion }, decisionDigest: decisionId, applicationState: "applied" }); + await db.update(issues).set({ lastStatusDecisionId: decisionId }).where(eq(issues.id, issueId)); + await db.insert(issueThreadInteractions).values({ id: interactionId, companyId, issueId, kind: "request_confirmation", sourceRunId, addresseeAgentId: agentId, createdByAgentId: workerId, + continuationPolicy: "wake_assignee", + status: scenario === "resolved-review" ? "accepted" : "pending", + ...(scenario === "resolved-review" ? { resolvedByRunId: runId, resolvedByAgentId: agentId } : {}), + payload: { version: 1, prompt: "Review the completed work.", target: { type: "custom", key: "native_completion_review", revisionId: decisionId } }, + }); + const contextSnapshot = { issueId, wakeReason: "native_completion_review", nativeReviewInteractionId: interactionId, nativeReviewDecisionId: decisionId }; + await db.update(heartbeatRuns).set({ contextSnapshot }).where(eq(heartbeatRuns.id, runId)); + if (scenario === "superseded-review") await db.update(issues).set({ lastStatusDecisionId: null }).where(eq(issues.id, issueId)); + const review = await getNativeReviewAssignment(db, { companyId, issueId, agentId, contextSnapshot, allowResolvedByRunId: runId }); + if (scenario === "superseded-review") expect(review).toBeNull(); + else expect(review?.interaction.status).toBe(scenario === "resolved-review" ? "accepted" : "pending"); + } + const event: PrpEvent = { + schema: "paperclip.prp.event.v1", runId, normalizedSessionId: native.normalizedSessionId, + turnId: native.turnId, sourceInstanceId: native.runnerInstanceId, sourceEventId: "model-terminal", sourceSeq: 1, + sourceKind: "runner", eventType: "turn.failed", schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), + payload: { status: "failed", error: { codexErrorInfo: "other", message: JSON.stringify({ + type: "error", status: 400, error: { type: "invalid_request_error", message: scenario === "unknown-error" ? "Private unknown failure" : `The '${model}' model is not supported when using Codex with a ChatGPT account.` }, + }) } }, + }; + const openInput = { identity: { companyId, issueId, runId, agentId, sessionId: native.normalizedSessionId }, backendKind: "mock" as const, sourceInstanceId: native.runnerInstanceId }; + if (scenario === "duplicate") { + const prior = new PaperclipControlPlanePort(db, { companyId, issueId, runId, agentId, sessionId: native.normalizedSessionId, sourceInstanceId: native.runnerInstanceId, controlPlaneSourceInstanceId: "prior-controller", completionContractId: contract.id, completionContractSha256: contract.canonicalSha256 }); + await prior.openRun(openInput); + expect((await prior.appendEvent(event)).disposition).toBe("committed"); + } + const turnId = scenario === "different-turn" ? "later-turn" : native.turnId; + const terminal: PrpTerminalState = { + schema: "paperclip.prp.terminal.v1", turnTerminalState: scenario === "succeeded" ? "completed" : scenario === "cancelled" ? "cancelled" : "failed", + runTerminalState: scenario === "succeeded" ? "succeeded" : scenario === "cancelled" ? "cancelled" : "failed", reportedWorkDisposition: "blocked", + }; + const result = { ...CONTROL_PLANE_CONFORMANCE_RESULT, reportedWorkDisposition: "blocked" as const, + blocker: { reasonCode: "provider_rejected", owner: { kind: "user" as const, name: "Task owner" }, unblockAction: "Select a compatible connection.", scope: "task_wide" as const }, + completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: completionInput.revision } }; + provider.execute.mockReset().mockImplementation(async (options: { controlPlane: ControlPlanePort }) => { + await options.controlPlane.openRun(openInput); + if (scenario === "untrusted-source") { + await expect(options.controlPlane.appendEvent({ ...event, sourceInstanceId: "wrong-runner" })).rejects.toThrow("native_event_source_binding_mismatch"); + } else if (scenario === "untrusted-run" || scenario === "untrusted-session") { + await expect(options.controlPlane.appendEvent({ ...event, + ...(scenario === "untrusted-run" ? { runId: randomUUID() } : { normalizedSessionId: "wrong-session" }), + })).rejects.toThrow("native_event_binding_mismatch"); + } else { + expect((await options.controlPlane.appendEvent(event)).disposition).toBe(scenario === "duplicate" ? "duplicate" : "committed"); + } + await options.controlPlane.completeRun({ result, terminal, turnId }); + return { result, terminal, turnId, normalizedSessionId: native.normalizedSessionId, providerSessionId: "fixture-provider-session", driverKind: "codex_app_server", driverVersion: "fixture", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + const adapter = await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, backend: { + descriptor: async () => { throw new Error("Unexpected provider call"); }, + openSession: async () => { throw new Error("Unexpected provider call"); }, + } }); + const classified = scenario === "committed" || scenario === "duplicate" || scenario.endsWith("-review") || scenario === "reassigned" || scenario === "successor-race"; + const reviewAuthorityEnded = scenario === "superseded-review" || scenario === "resolved-review"; + const ownerChanged = scenario === "reassigned" || scenario === "successor-race"; + expect(adapter.errorCode).toBe(classified ? "native_provider_model_rejected" : undefined); + expect(adapter.errorMessage).toBe(classified ? NATIVE_MODEL_REJECTION_MESSAGE : scenario === "succeeded" ? null : `Native session ${terminal.runTerminalState}`); + expect(adapter.resultJson?.nativeProviderFailure).toEqual(classified ? NATIVE_MODEL_REJECTION_DIAGNOSTIC : undefined); + expect(adapter.nativeFinalization).toMatchObject({ result, terminal, turnId, workspaceFinalizeStatus: "pending" }); + expect(provider.execute).toHaveBeenCalledTimes(1); + expect(execution.provider).toEqual({ kind: "codex", model }); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + expect(coordinator).toMatchObject({ phase: "workspace_finalizing", leaseOwner: null, attempt: 1, nextAttemptAt: null }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toHaveLength(1); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId))).toHaveLength(scenario.startsWith("untrusted-") ? 0 : 1); + const diagnostics = collectRunFailureDiagnostics({ ...run, resultJson: adapter.resultJson ?? null }, {}); + expect(diagnostics.provider).toEqual(classified ? NATIVE_MODEL_REJECTION_DIAGNOSTIC : {}); + expect(JSON.stringify(diagnostics)).not.toContain(model); + expect(JSON.stringify(diagnostics)).not.toContain("Private"); + const [persistedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(persistedRun.errorCode).toBeNull(); // No adapter metadata has been saved yet. + expect(await readPersistedNativeModelRejection(db, persistedRun, turnId, terminal)).toEqual(classified ? { + errorCode: "native_provider_model_rejected", errorMessage: NATIVE_MODEL_REJECTION_MESSAGE, diagnostic: NATIVE_MODEL_REJECTION_DIAGNOSTIC, + } : null); + if (classified) { + if (scenario === "committed") { + // Neither saved display diagnostics nor a different execution binding + // can confer the right to change the finalizer's retry decision. + for (const changed of [ + { ...persistedRun, runnerProfileJson: null }, + { ...persistedRun, nativeSessionId: "foreign-session" }, + { ...persistedRun, runnerInstanceId: "foreign-source" }, + { ...persistedRun, companyId: randomUUID() }, + { ...persistedRun, completionContractSha256: "foreign-contract" }, + { ...persistedRun, runnerProfileJson: { nativeExecutionInput: { ...execution, provider: { kind: "codex", model: "different-model" } } } }, + { ...persistedRun, runnerProfileJson: { nativeExecutionInput: { ...execution, binding: { ...execution.binding, runId: randomUUID() } } } }, + ]) expect(await readPersistedNativeModelRejection(db, changed, turnId, terminal)).toBeNull(); + const [stored] = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); + await db.update(heartbeatRunEvents).set({ sourcePayloadSha256: "invalid-hash" }).where(eq(heartbeatRunEvents.id, stored.id)); + expect(await readPersistedNativeModelRejection(db, persistedRun, turnId, terminal)).toBeNull(); + await db.update(heartbeatRunEvents).set({ sourcePayloadSha256: stored.sourcePayloadSha256 }).where(eq(heartbeatRunEvents.id, stored.id)); + } + // Simulate a restart after native result acceptance, before heartbeat's + // adapter update. Finalization must use durable provider evidence. + const acceptedBefore = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId)); + if (scenario === "reassigned") { + await db.update(issues).set({ assigneeAgentId: workerId, executionRunId: null }).where(eq(issues.id, issueId)); + } else if (scenario === "successor-race") { + const [snapshot] = await db.select().from(issues).where(eq(issues.id, issueId)); + const newerRunId = randomUUID(), raceAssessmentId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: newerRunId, companyId, agentId, status: "running", nativeIssueId: issueId }); + await db.update(issues).set({ executionRunId: newerRunId }).where(eq(issues.id, issueId)); + const [changed] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(changed.statusVersion).toBe(snapshot.statusVersion); + await db.insert(workAssessments).values({ id: raceAssessmentId, companyId, issueId, runId, contractId: contract.id, resultId: acceptedBefore[0].id, triggerKind: "native_result", triggerActorCompanyId: companyId, priorIssueStatus: snapshot.status, priorStatusVersion: snapshot.statusVersion, policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, assessmentJson: {}, inputDigest: raceAssessmentId }); + await expect(commitNativeStatusDecision({ db, companyId, issueId, runId, assessmentId: raceAssessmentId, + priorStatus: snapshot.status, priorStatusVersion: snapshot.statusVersion, priorDecisionId: snapshot.lastStatusDecisionId, + requireModelRejectionOwner: { agentId, reviewContext: null }, + decision: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "blocked", toStatus: "blocked", reasonCode: "native_provider_model_rejected", unblockDescriptor: { owner: "board", action: NATIVE_MODEL_REJECTION_MESSAGE }, effects: [{ kind: "bind_blocker", owner: "board", action: NATIVE_MODEL_REJECTION_MESSAGE }] }, + })).rejects.toBeInstanceOf(NativeStatusRaceError); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId))).toHaveLength(0); + } + for (let retry = 0; retry < 2; retry++) { + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + } + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toEqual(acceptedBefore); + const [finalRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(finalRun).toMatchObject({ status: "failed", errorCode: "native_provider_model_rejected", error: NATIVE_MODEL_REJECTION_MESSAGE }); + const [finalIssue] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(finalIssue.status).toBe(scenario.endsWith("-review") ? "in_review" : ownerChanged ? "in_progress" : "blocked"); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId)); + expect(decisions).toHaveLength(1); + expect(decisions[0].reasonCode).toBe(reviewAuthorityEnded ? "native_review_action_finished" : ownerChanged ? "run_failed_partial_evidence_preserved" : "native_provider_model_rejected"); + const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + expect(wakes.filter(wake => wake.payload?.issueId === issueId)).toHaveLength(ownerChanged ? 1 : 0); + if (!ownerChanged) expect(await heartbeatService(db).dispatchPendingNativeStatusWakeups({ companyId })).toEqual({ scanned: 0, dispatched: 0, recovered: 0, deferred: 0 }); + expect(provider.execute).toHaveBeenCalledTimes(1); + if (scenario === "pending-review") { + const reviewContext = readNativeReviewAssignmentContext(finalRun.contextSnapshot)!; + expect(finalIssue).toMatchObject({ lastStatusDecisionId: reviewContext.nativeReviewDecisionId, executionRunId: null, checkoutRunId: null }); + const originalReview = await getNativeReviewAssignment(db, { companyId, issueId, agentId, contextSnapshot: reviewContext }); + expect(originalReview?.interaction.status).toBe("pending"); + expect(finalIssue.statusVersion).toBe(originalReview!.sourceDecision.decisionJson.projectedStatusVersion); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + + // Exercise the real release and periodic recovery lanes, rather than + // assuming that absence of a finalizer wake prevents later retries. + const unexpected = vi.fn(async () => { throw new Error("Unexpected automatic recovery"); }); + const release = createReleaseIssueExecution({ + issueLock: createPostgresWakeQueueAdapter(db, { resolveResponsibleUserId: unexpected, getRoutineEnv: unexpected, resolveSessionBeforeForWakeup: unexpected }), + recovery: { escalateStrandedAssignedIssue: unexpected, escalateStrandedRecoveryIssueInPlace: unexpected }, + }); + expect((await release({ companyId, runId, now: new Date() })).outcome.kind).toBe("released"); + await heartbeatService(db).reconcileStrandedAssignedIssues(); + expect(unexpected).not.toHaveBeenCalled(); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId))).filter(wake => wake.payload?.issueId === issueId)).toHaveLength(0); + const [afterRecovery] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(afterRecovery).toMatchObject({ status: "in_review", statusVersion: finalIssue.statusVersion, lastStatusDecisionId: finalIssue.lastStatusDecisionId }); + + // The operator repairs configuration and explicitly retries the + // assigned reviewer. No review rebinding or status reset is needed. + await db.update(agents).set({ adapterConfig: { provider: "codex", model: "repaired-model" }, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }).where(eq(agents.id, agentId)); + // Occupy the execution slot to exercise real wake admission without + // starting a provider. The normal atomic review claim follows below. + const occupiedRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: occupiedRunId, companyId, agentId, status: "running", startedAt: new Date() }); + const retryRun = await heartbeatService(db, { runtimeEnv: {} }).wakeup(agentId, { + failedRunId: runId, source: "on_demand", triggerDetail: "manual", reason: "retry_failed_run", + requestedByActorType: "user", requestedByActorId: "review-repair-operator", + payload: { issueId }, contextSnapshot: { ...reviewContext }, + }); + expect(retryRun).toMatchObject({ status: "queued", retryOfRunId: runId, contextSnapshot: { issueId, ...reviewContext } }); + const retryRunId = retryRun!.id; + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, occupiedRunId)); + expect(await claimQueuedNativeReviewRun(db, { run: retryRun!, agentNameKey: "reviewer", claimedAt: new Date(), claimValues: { nativeIssueId: issueId, runtimeMode: "native" } })) + .toMatchObject({ id: retryRunId, status: "running" }); + expect(await getNativeReviewAssignment(db, { companyId, issueId, agentId, contextSnapshot: reviewContext, actingRunId: retryRunId, issueExecutionRunId: retryRunId })).not.toBeNull(); + await issueThreadInteractionService(db).acceptInteraction( + { id: issueId, companyId, projectId: null, goalId: null, status: "in_review" }, + reviewContext.nativeReviewInteractionId, {}, { agentId, runId: retryRunId }, + ); + const [resolved] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, reviewContext.nativeReviewInteractionId)); + expect(resolved).toMatchObject({ status: "accepted", resolvedByAgentId: agentId, resolvedByRunId: retryRunId }); + expect(await db.select().from(issues).where(eq(issues.id, issueId))).toMatchObject([{ status: "done", assigneeAgentId: workerId }]); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).toEqual(acceptedBefore); + expect(provider.execute).toHaveBeenCalledTimes(1); + await db.update(agents).set({ adapterConfig: { provider: "codex", model } }).where(eq(agents.id, agentId)); + } + } else if (scenario === "unknown-error") { + // A lookalike diagnostic flag does not alter unknown failed-result policy. + await db.update(heartbeatRuns).set({ resultJson: { nativeProviderFailure: NATIVE_MODEL_REJECTION_DIAGNOSTIC } }).where(eq(heartbeatRuns.id, runId)); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" }); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, runId)); + expect(decision.reasonCode).toBe("run_failed_partial_evidence_preserved"); + const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + expect(wakes.filter(wake => wake.payload?.issueId === issueId)).toHaveLength(1); + } + }, + ); +}); diff --git a/server/src/services/native-runtime/native-provider-failure.test.ts b/server/src/services/native-runtime/native-provider-failure.test.ts new file mode 100644 index 00000000000..5db2942e4c2 --- /dev/null +++ b/server/src/services/native-runtime/native-provider-failure.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; +import type { PrpEvent, PrpTerminalState } from "../../vendor/paperclip-runner/index.js"; +import { createNativeProviderFailureObservation, NATIVE_MODEL_REJECTION_DIAGNOSTIC, NATIVE_MODEL_REJECTION_MESSAGE } from "./native-provider-failure.js"; + +const model = "test-model"; +const response = { + type: "error", status: 400, + error: { type: "invalid_request_error", message: `The '${model}' model is not supported when using Codex with a ChatGPT account.` }, +}; +const event: PrpEvent = { + schema: "paperclip.prp.event.v1", runId: "run", normalizedSessionId: "session", turnId: "turn", + sourceInstanceId: "runner", sourceEventId: "failure", sourceSeq: 1, sourceKind: "runner", + eventType: "turn.failed", schemaVersion: 1, priority: 0, emittedAt: "2026-01-01T00:00:00.000Z", + payload: { status: "failed", error: { message: JSON.stringify(response), codexErrorInfo: "other" } }, +}; +const terminal: PrpTerminalState = { + schema: "paperclip.prp.terminal.v1", turnTerminalState: "failed", runTerminalState: "failed", reportedWorkDisposition: "blocked", +}; +const observation = () => createNativeProviderFailureObservation({ kind: "codex", model }); + +describe("native provider failure observation", () => { + it("binds the structured rejection to the failed turn and emits no response or model text", () => { + const observed = observation(); + observed.observe(event); + observed.observe(event); // A byte-identical committed replay remains idempotent. + const result = observed.forTerminal("turn", terminal); + expect(result).toEqual({ errorCode: "native_provider_model_rejected", errorMessage: NATIVE_MODEL_REJECTION_MESSAGE, diagnostic: NATIVE_MODEL_REJECTION_DIAGNOSTIC }); + expect(JSON.stringify(result)).not.toContain(model); + expect(JSON.stringify(result)).not.toContain("invalid_request_error"); + expect(observed.forTerminal("later-turn", terminal)).toBeNull(); + expect(observed.forTerminal(null, terminal)).toBeNull(); + expect(observed.forTerminal("turn", { ...terminal, runTerminalState: "succeeded" })).toBeNull(); + expect(observed.forTerminal("turn", { ...terminal, turnTerminalState: "cancelled" })).toBeNull(); + }); + + it.each([ + { ...event, sourceKind: "control_plane" as const }, + { ...event, eventType: "item.completed" as const }, + { ...event, eventType: "harness.diagnostic" as const }, + { ...event, turnId: undefined }, + { ...event, payload: { ...event.payload, status: "completed" } }, + { ...event, payload: { ...event.payload, recoverable: true } }, + { ...event, payload: { status: "failed", error: { message: JSON.stringify(response), recoverable: true } } }, + ])("ignores non-authoritative or retryable evidence %#", (unrelated) => { + const observed = observation(); + observed.observe(unrelated); + expect(observed.forTerminal("turn", terminal)).toBeNull(); + }); + + it.each([ + response.error.message, "malformed JSON", " ".repeat(4_097), + JSON.stringify({ ...response, status: 401 }), + JSON.stringify({ ...response, status: "400" }), + JSON.stringify({ ...response, type: "notice" }), + JSON.stringify({ ...response, error: { ...response.error, type: "server_error" } }), + JSON.stringify({ ...response, error: { ...response.error, message: "Private arbitrary provider text" } }), + JSON.stringify({ ...response, error: { ...response.error, message: response.error.message + " Private suffix" } }), + JSON.stringify({ ...response, error: { ...response.error, message: response.error.message.replace(model, "another-model") } }), + JSON.stringify([response]), "null", + ])("leaves unknown failures unchanged %#", (message) => { + const observed = observation(); + observed.observe({ ...event, payload: { status: "failed", error: { message } } }); + expect(observed.forTerminal("turn", terminal)).toBeNull(); + }); + + it.each([{ kind: "claude", model }, { kind: "codex" }, { kind: "codex", model: "other" }])( + "does not apply a rejection to another selected provider/model (%j)", (provider) => { + const observed = createNativeProviderFailureObservation(provider); + observed.observe(event); + expect(observed.forTerminal("turn", terminal)).toBeNull(); + }, + ); +}); diff --git a/server/src/services/native-runtime/native-provider-failure.ts b/server/src/services/native-runtime/native-provider-failure.ts new file mode 100644 index 00000000000..ec7380ffc86 --- /dev/null +++ b/server/src/services/native-runtime/native-provider-failure.ts @@ -0,0 +1,60 @@ +import type { PrpEvent, PrpTerminalState } from "../../vendor/paperclip-runner/index.js"; + +export const NATIVE_MODEL_REJECTION_MESSAGE = + "The selected model is not supported by the current ChatGPT connection. Choose a supported model or a compatible AI connection."; + +export const NATIVE_MODEL_REJECTION_DIAGNOSTIC = { + provider: "codex", + category: "model_auth_incompatible", + status: 400, + authMode: "chatgpt", +} as const; + +function record(value: unknown): Record { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? value as Record + : {}; +} + +/** Reconstruct only the closed diagnostic vocabulary, never provider text. */ +export function readNativeModelRejectionDiagnostic(value: unknown) { + try { + const entry = record(value); + return Object.entries(NATIVE_MODEL_REJECTION_DIAGNOSTIC).every(([key, expected]) => entry[key] === expected) + ? { ...NATIVE_MODEL_REJECTION_DIAGNOSTIC } + : null; + } catch { return null; } +} + +/** Observe only events accepted by the bound control-plane persistence port. */ +export function createNativeProviderFailureObservation(provider: { kind: string; model?: string | null }) { + let rejectedTurnId: string | null = null; + return { + observe(event: PrpEvent) { + if (provider.kind !== "codex" || !provider.model || + event.sourceKind !== "runner" || event.eventType !== "turn.failed" || !event.turnId) return; + const payload = record(event.payload); + const failure = record(payload.error); + if (payload.status !== "failed" || payload.recoverable === true || failure.recoverable === true || + typeof failure.message !== "string" || failure.message.length > 4_096) return; + let response: Record; + try { response = record(JSON.parse(failure.message)); } catch { return; } + const error = record(response.error); + // This is a provider terminal envelope, not text emitted by a tool or an + // agent. Bind the exact rejection to the configured model without storing + // its name or the response. Unknown provider errors retain their behavior. + if (response.type !== "error" || response.status !== 400 || error.type !== "invalid_request_error" || + error.message !== `The '${provider.model}' model is not supported when using Codex with a ChatGPT account.`) return; + rejectedTurnId = event.turnId; + }, + forTerminal(turnId: string | null | undefined, terminal: PrpTerminalState) { + if (!rejectedTurnId || turnId !== rejectedTurnId || + terminal.turnTerminalState !== "failed" || terminal.runTerminalState !== "failed") return null; + return { + errorCode: "native_provider_model_rejected" as const, + errorMessage: NATIVE_MODEL_REJECTION_MESSAGE, + diagnostic: { ...NATIVE_MODEL_REJECTION_DIAGNOSTIC }, + }; + }, + }; +} diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index b6aea881b4e..fb248551b92 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,4 +1,5 @@ import { settleSlackConversation } from "../slack-conversation-lifecycle.js"; +import { readPersistedNativeModelRejection } from "./native-provider-failure-evidence.js"; import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js"; import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; import { conversationNativeDecision, isConversation } from "../agent-conversations.js"; @@ -24,7 +25,7 @@ import { workspaceOperations, } from "@paperclipai/db"; import { classifyNativeEvidence } from "./evidence-classifier.js"; -import type { PrpIgnoredAttentionRequest } from "@paperclipai/paperclip-runner"; +import type { PrpIgnoredAttentionRequest, PrpTerminalState } from "@paperclipai/paperclip-runner"; import { arbitrateNativeStatus, NATIVE_STATUS_ARBITER_POLICY_VERSION, @@ -1275,7 +1276,15 @@ export async function finalizeNativeRun(input: { companyId: run.companyId, issueId: authoritativeIssue.id, agentId: run.agentId, contextSnapshot: reviewContext, allowResolvedByRunId: run.id, }) : null; + const providerModelRejection = await readPersistedNativeModelRejection( + input.db, run, resultRow.turnId, envelope.terminal as PrpTerminalState, + ); + const ownsModelRejectionDecision = + (!authoritativeIssue.executionRunId || authoritativeIssue.executionRunId === run.id) && + (reviewContext ? nativeReview?.interaction.status === "pending" + : authoritativeIssue.assigneeAgentId === run.agentId && !authoritativeIssue.assigneeUserId); const proposedDecision = resolveNativeFinalizerStatus({ + providerModelRejected: providerModelRejection !== null && ownsModelRejectionDecision, ...(reviewContext ? { nativeReviewOutcome: nativeReview ? nativeReview.interaction.status === "pending" ? "pending" as const : "resolved" as const : "stale" as const } : {}), @@ -1358,6 +1367,9 @@ export async function finalizeNativeRun(input: { priorStatusVersion: Number(authoritativeIssue.statusVersion), priorDecisionId: authoritativeIssue.lastStatusDecisionId, decision, + requireModelRejectionOwner: decision.reasonCode === "native_provider_model_rejected" + ? { agentId: run.agentId, reviewContext } + : undefined, requireBoardResponseWaitSource: decision.reasonCode === "board_response_waiting" || repairBoardResponseWait ? boardResponseWait?.source @@ -1400,6 +1412,7 @@ export async function finalizeNativeRun(input: { ); const clearExecutionFailure = input.projectRunStatus && finalizationPhase === "committed" && terminalState === "succeeded"; const finalizationMetadata = { + ...(providerModelRejection ? { nativeProviderFailure: providerModelRejection.diagnostic } : {}), finalizationPhase, ...(finalizationPhase === "committed" ? { failureCode: null, originalFailureCode: null, nextAttemptAt: null } : {}), assessmentId: assessmentRow.id, @@ -1450,6 +1463,12 @@ export async function finalizeNativeRun(input: { nativePhase: finalizationPhase, nativePhaseUpdatedAt: now, ...(clearExecutionFailure ? { error: null, errorCode: null } : {}), + ...(providerModelRejection ? { + // Recovery can finalize before heartbeat saves the adapter result. + // Retain any independently recorded execution/cleanup failure. + error: sql`coalesce(${heartbeatRuns.error}, ${providerModelRejection.errorMessage})`, + errorCode: sql`coalesce(${heartbeatRuns.errorCode}, ${providerModelRejection.errorCode})`, + } : {}), resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(finalizationMetadata)}::jsonb || ${clearExecutionFailure ? recoveredExecutionFailureMetadata() : sql`'{}'::jsonb`}`, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index cc217b94525..ec28b68800e 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9,6 +9,7 @@ import { REMOTE_CODEX_SUPPORTED_RANGE, } from "./codex-runtime-compatibility.js"; import { createNativeToolTrace, type NativeToolTrace } from "./native-tool-trace.js"; +import { createNativeProviderFailureObservation } from "./native-provider-failure.js"; import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js"; import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js"; import { bindManagedNativeCredentialTurn, completeManagedNativeCredentialTurn } from "./managed-native-credentials.js"; @@ -7805,6 +7806,7 @@ async function executePaperclipNativeSessionWithinScope( let turnStartedAtMs: number | null = null; let firstAgentEventRecorded = false; let providerUsageLimitObserved = false; + const providerFailureObservation = createNativeProviderFailureObservation(input.execution.provider); let turnCompletedAtMs: number | null = null; let runnerSessionStartupScope: NativeRunSpanScope | null = null; let agentTurnScope: NativeRunSpanScope | null = null; @@ -7861,6 +7863,7 @@ async function executePaperclipNativeSessionWithinScope( await liveQuestions.observe(event); await projectSessionGoalEvent(event); providerUsageLimitObserved ||= nativeProviderUsageLimitFromEvent(event); + providerFailureObservation.observe(event); const eventAtMs = Date.parse(event.emittedAt); const milestoneAtMs = Number.isFinite(eventAtMs) ? eventAtMs @@ -8064,6 +8067,7 @@ async function executePaperclipNativeSessionWithinScope( await liveQuestions.observe(event); await projectSessionGoalEvent(event); providerUsageLimitObserved ||= nativeProviderUsageLimitFromEvent(event); + providerFailureObservation.observe(event); const questionFallback = await materializeRuntimeQuestionFallback({ db: input.db, binding: input.execution.binding, @@ -9203,6 +9207,7 @@ async function executePaperclipNativeSessionWithinScope( ); if (collectInstructions && !collectedByOwner) await instructionCopy!.collectStopped(); } + const providerFailure = providerFailureObservation.forTerminal(native.turnId, native.terminal); const adapterResult: AdapterExecutionResult = { exitCode: native.terminal.runTerminalState === "succeeded" ? 0 : 1, signal: null, @@ -9210,10 +9215,12 @@ async function executePaperclipNativeSessionWithinScope( errorMessage: native.terminal.runTerminalState === "succeeded" ? null - : `Native session ${native.terminal.runTerminalState}`, + : providerFailure?.errorMessage ?? `Native session ${native.terminal.runTerminalState}`, + ...(providerFailure ? { errorCode: providerFailure.errorCode } : {}), resultJson: { nativeResult: native.result as unknown as Record, nativeTerminal: native.terminal as unknown as Record, + ...(providerFailure ? { nativeProviderFailure: providerFailure.diagnostic } : {}), ...(native.goalRolloverRequired ? { goalRolloverRequired: true } : {}), planSynchronizations, }, diff --git a/server/src/services/native-runtime/status-arbiter.test.ts b/server/src/services/native-runtime/status-arbiter.test.ts index 0f3b1abad9c..41625482173 100644 --- a/server/src/services/native-runtime/status-arbiter.test.ts +++ b/server/src/services/native-runtime/status-arbiter.test.ts @@ -44,6 +44,23 @@ function arbitrate( } describe("native status authority", () => { + it.each(["in_progress", "in_review"] as const)("blocks a current worker's proven model rejection without a retry (%s)", (priorIssueStatus) => { + const decision = arbitrate({ priorIssueStatus, terminalState: "failed", providerModelRejected: true }); + expect(decision).toMatchObject({ statusAction: "blocked", toStatus: "blocked", reasonCode: "native_provider_model_rejected", unblockDescriptor: { owner: "board" } }); + expect(decision.effects).toEqual([{ kind: "bind_blocker", owner: "board", action: expect.any(String) }]); + expect(arbitrate({ terminalState: "failed", providerModelRejected: false }).effects).toContainEqual(expect.objectContaining({ kind: "schedule_retry" })); + expect(arbitrate({ terminalState: "succeeded", providerModelRejected: true }).reasonCode).not.toBe("native_provider_model_rejected"); + expect(arbitrate({ terminalState: "failed", providerModelRejected: true, workspaceFinalizeStatus: "failed" }).reasonCode).toBe("finalization_failed_claim_preserved"); + expect(arbitrate({ terminalState: "failed", providerModelRejected: true, priorIssueStatus: "done" }).toStatus).toBe("done"); + for (const nativeReviewOutcome of ["stale", "resolved"] as const) { + expect(arbitrate({ terminalState: "failed", providerModelRejected: true, priorIssueStatus, nativeReviewOutcome })) + .toMatchObject({ statusAction: "preserve", toStatus: priorIssueStatus, reasonCode: "native_review_action_finished" }); + } + }); + it("preserves pending review authority without automatic recovery after model rejection", () => { + expect(arbitrate({ priorIssueStatus: "in_review", terminalState: "failed", providerModelRejected: true, nativeReviewOutcome: "pending" })) + .toMatchObject({ statusAction: "preserve", toStatus: "in_review", reasonCode: "native_provider_model_rejected", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] }); + }); it("a reviewer finishes its decision without completing rejected or still-reviewed work", () => { for (const priorIssueStatus of ["in_progress", "in_review"] as const) { const decision = arbitrate({ priorIssueStatus, nativeReviewOutcome: "resolved" }); @@ -511,7 +528,7 @@ describe("native status authority", () => { expect.objectContaining({ statusAction: "blocked", toStatus: "blocked", - policyVersion: "phase6-v7", + policyVersion: "phase6-v8", reasonCode: "current_track_blocker_waiting", unblockDescriptor: { owner: "board", diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index 55e298bf5aa..c4235536a6b 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -1,6 +1,7 @@ import type { NativeEvidenceAssessment } from "./evidence-classifier.js"; +import { NATIVE_MODEL_REJECTION_MESSAGE } from "./native-provider-failure.js"; -export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v7"; +export const NATIVE_STATUS_ARBITER_POLICY_VERSION = "phase6-v8"; export type NativeAuthoritativeIssueStatus = | "backlog" @@ -115,6 +116,8 @@ export function arbitrateNativeStatus(input: { reviewOwnerUserId?: string | null; /** Review decisions own task state; a reviewer's finish report cannot override them. */ nativeReviewOutcome?: "resolved" | "pending" | "stale"; + /** Re-derived from the committed runner terminal and pinned execution identity. */ + providerModelRejected?: boolean; agentId: string; priorIssueStatus: NativeAuthoritativeIssueStatus; }): NativeStatusDecision { @@ -146,6 +149,31 @@ export function arbitrateNativeStatus(input: { ], }; } + if (input.providerModelRejected && input.terminalState === "failed" && + input.nativeReviewOutcome !== "stale" && input.nativeReviewOutcome !== "resolved") { + if (input.nativeReviewOutcome === "pending") { + // The pending review is bound to the worker's status decision and + // version. Preserve that authority so an explicit retry after a + // configuration repair can resolve the same review. In particular, + // do not create the normal unresolved-review recovery action. + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode: "native_provider_model_rejected", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }], + }; + } + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "blocked", + toStatus: "blocked", + reasonCode: "native_provider_model_rejected", + unblockDescriptor: { owner: "board", action: NATIVE_MODEL_REJECTION_MESSAGE }, + effects: [{ kind: "bind_blocker", owner: "board", action: NATIVE_MODEL_REJECTION_MESSAGE }], + }; + } if (input.nativeReviewOutcome) { return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 307efb532dc..21c14f5e934 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -24,6 +24,7 @@ import { type NativeStatusEffect, } from "./status-arbiter.js"; import { nativeSha256 } from "./canonical.js"; +import { getNativeReviewAssignment, type NativeReviewAssignmentContext } from "./native-review-participant.js"; import { readNativeBoardResponseWaitSource, readNativeBoardResponseWaitOrigin, @@ -1535,6 +1536,7 @@ export async function commitNativeStatusDecision(input: { preMaterializedEffects?: NativeMaterializedStatusEffect[]; supersedesCommittedDecisionId?: string; requireExternalChatResponseWaitAuthorization?: { agentId: string }; + requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; requireBoardResponseWaitSource?: NativeBoardResponseWaitSource; requireBoardResponseWaitOrigin?: NativeBoardResponseWaitOrigin; reviewResponsePresentation?: { @@ -1547,6 +1549,9 @@ export async function commitNativeStatusDecision(input: { throw new Error("native_status_reason_code_required"); } const reasonCode = input.decision.reasonCode; + if (reasonCode === "native_provider_model_rejected" && !input.requireModelRejectionOwner) { + throw new Error("native_model_rejection_owner_required"); + } const publications: ActivityPublication[] = []; const terminalRunsToEmit: (typeof heartbeatRuns.$inferSelect)[] = []; const committed = await input.db.transaction(async (tx) => { @@ -1616,6 +1621,20 @@ export async function commitNativeStatusDecision(input: { ) { throw new NativeStatusRaceError(); } + if (input.requireModelRejectionOwner) { + const owner = input.requireModelRejectionOwner; + // A successor can claim execution without changing statusVersion. Keep + // this new blocking authority behind the same locked issue snapshot. + if (issue.executionRunId && issue.executionRunId !== input.runId) throw new NativeStatusRaceError(); + if (owner.reviewContext) { + const review = await getNativeReviewAssignment(tx as unknown as Db, { + companyId: input.companyId, issueId: input.issueId, agentId: owner.agentId, contextSnapshot: owner.reviewContext, + }); + if (review?.interaction.status !== "pending") throw new NativeStatusRaceError(); + } else if (issue.assigneeAgentId !== owner.agentId || issue.assigneeUserId) { + throw new NativeStatusRaceError(); + } + } if (input.requireExternalChatResponseWaitAuthorization) { let authorization; try { diff --git a/server/src/services/recovery/provider-failure-classification.test.ts b/server/src/services/recovery/provider-failure-classification.test.ts index d58bcf42a9f..9343e878138 100644 --- a/server/src/services/recovery/provider-failure-classification.test.ts +++ b/server/src/services/recovery/provider-failure-classification.test.ts @@ -7,8 +7,20 @@ import { classifyContinuationFailure, } from "./service.js"; import { legacyExecutionNeedsReconciliation } from "../legacy-execution-recovery.js"; +import { isConfigurationIncompleteFailedRun } from "../../modules/wake-queue/domain/values.js"; describe("classifyAdapterFailureForRecovery", () => { + it("waits for model configuration changes instead of retrying a typed native model rejection", () => { + const run = { errorCode: "native_provider_model_rejected", error: "Fixed model configuration message", resultJson: null }; + expect(classifyAdapterFailureForRecovery(run)).toEqual({ kind: "configuration_incomplete" }); + expect(classifyContinuationFailure(run as never)).toMatchObject({ kind: "non_retryable", maxAttempts: 0 }); + expect(isConfigurationIncompleteFailedRun(run)).toBe(true); + expect(isAiAuthenticationBlocked(run as never)).toBe(false); + const unknown = { ...run, errorCode: "adapter_failed", error: "native_provider_model_rejected" }; + expect(classifyAdapterFailureForRecovery(unknown)).toBeNull(); + expect(classifyContinuationFailure(unknown as never).kind).toBe("transient_infra"); + expect(isConfigurationIncompleteFailedRun(unknown)).toBe(false); + }); it.each(["acpx_auth_required", "claude_auth_required", "codex_auth_required", "adapter_auth_missing", "refresh_token_reused", "authentication_required"])("does not automatically retry provider authentication failure %s", (errorCode) => { const classification = classifyRunLiveness({ runStatus: "failed", issue: null, errorCode, authenticationRepairRequested: true }); const run = { errorCode, ...classification }; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index cfda20e5bc5..2cb7ac951f6 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -491,6 +491,7 @@ const TRANSIENT_INFRA_CONTINUATION_ERROR_CODES = new Set([ ]); const NON_RETRYABLE_CONTINUATION_ERROR_CODES = new Set([ + "native_provider_model_rejected", "provider_tool_definition_invalid", "adapter_engine_unavailable", "agent_not_invokable", @@ -626,7 +627,7 @@ export function classifyAdapterFailureForRecovery( ): AdapterFailureRecoveryClassification { // An engine prerequisite cannot be repaired by asking the same unavailable // engine to retry. Use the existing configuration-blocker path. - if (latestRun.errorCode === "adapter_engine_unavailable" || latestRun.errorCode === "provider_tool_definition_invalid") { + if (latestRun.errorCode === "adapter_engine_unavailable" || latestRun.errorCode === "provider_tool_definition_invalid" || latestRun.errorCode === "native_provider_model_rejected") { return { kind: "configuration_incomplete" }; } if ( diff --git a/server/src/services/run-failure-diagnostics.ts b/server/src/services/run-failure-diagnostics.ts index 5a83450794b..c5c9d7e820b 100644 --- a/server/src/services/run-failure-diagnostics.ts +++ b/server/src/services/run-failure-diagnostics.ts @@ -5,6 +5,7 @@ import { redactDiagnosticText } from "@paperclipai/adapter-utils/command-redacti import { sanitizeWorkspaceRestoreDiagnostic } from "@paperclipai/adapter-utils/workspace-restore-diagnostics"; import { redactCurrentUserText } from "../log-redaction.js"; import { redactSensitiveText, REDACTED_EVENT_VALUE } from "../redaction.js"; +import { readNativeModelRejectionDiagnostic } from "./native-runtime/native-provider-failure.js"; type Run = typeof heartbeatRuns.$inferSelect; type Context = Record; @@ -184,6 +185,8 @@ export function collectRunFailureDiagnostics(run: Run, options: RunFailureReport "stackPreview", "status", "statusCode", "requestId", ]); const provider = scalars(read(result, "terminalSessionFailure"), ["category", "title", "details"]); + const nativeProviderFailure = readNativeModelRejectionDiagnostic(read(result, "nativeProviderFailure")); + if (nativeProviderFailure) Object.assign(provider, nativeProviderFailure); const truncatedFields: string[] = []; const providerTruncation = read(read(result, "terminalSessionFailure"), "truncatedFields"); if (Array.isArray(providerTruncation)) {