From bf12b2d8ade5215834071f941f929fec72bcb417 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 7 Nov 2026 02:14:42 -0800 Subject: [PATCH] fix: retry reviewed chat setup after temporary contention (#19787) Retry reviewed chat preparation with the original accepted input. Bound retries for authorization proof waits that expire before provider work. Preserve the original answer receipt and queue later wakes separately. Keep plan confirmation semantics by assigning the interaction marker only to question answer delivery and legacy question payloads. Verification: 433 tests across four backend suites, all 17 project-chat browser tests, and all server source/test typecheck layers passed. The plan approval regression was reproduced on the previous head before the fix. All applicable CI passed and exact-head Greptile Apex scored 5/5. Co-Authored-By: Paperclip --- doc/execution-semantics.md | 19 ++++ .../heartbeat-retry-scheduling.test.ts | 36 ++++++- ...-reviewed-chat-binding.integration.test.ts | 37 +++++++ .../src/modules/wake-queue/domain/context.ts | 10 +- server/src/modules/wake-queue/index.ts | 1 + server/src/services/heartbeat/queue.ts | 5 +- server/src/services/heartbeat/retries.test.ts | 36 +++++++ server/src/services/heartbeat/retries.ts | 4 +- .../services/heartbeat/run-completion.test.ts | 82 +++++++++++++++- .../src/services/heartbeat/run-completion.ts | 9 +- .../src/services/heartbeat/run-preparation.ts | 11 ++- .../external-chat-question-response-retry.ts | 52 ++++++++++ .../external-chat-question-response.ts | 25 ++--- .../external-chat-wait.integration.test.ts | 96 ++++++++++++++++++- .../services/question-response-delivery.ts | 1 + .../reviewed-chat-binding-diagnostics.ts | 9 ++ 16 files changed, 405 insertions(+), 28 deletions(-) create mode 100644 server/src/services/native-runtime/external-chat-question-response-retry.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index ae2d25e3485..9047d556370 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -302,6 +302,25 @@ Messages from the same conversation still coalesce into one run. Messages from different conversations queue separately and run one after the other. Wakes without a conversation keep their normal coalescing. +If reviewed chat attestation exhausts its short lock or delivery wait before +provider dispatch, the run records `reviewed_chat_execution_binding_not_ready`. +The existing durable retry scheduler permits at most two further attempts. Each +attempt retains the admitted comment batch and checks current execution ownership, +provider access and principal permissions again. An answered interaction retains +its own bounded continuation retry. A denied authorization, Stop, pause or blocked +dependency cannot use this wait as permission to execute. No retry budget resets, +new background poller or provider replay are introduced. + +Question-answer delivery records its interaction marker in the wake context. +Queue enrichment backfills that marker for older question-answer payloads only. +Plan confirmations retain their existing admission and handoff lifecycle. + +An answered retry proves at most two saved pre-dispatch failure ancestors before +reading the original user-owned answer receipt. It rechecks that receipt, its +original delivery target and current chat permissions. The original receipt and +delivery remain unchanged. New comments and control wakes queue separately from +a retained chat retry, including while the retry waits for its scheduled time. + For a truncated external-chat wake, native completion feedback requires the run-bound complete comment-read receipt before accepting a finish report. An unread, partial, stale or foreign receipt returns a corrective diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index dda93d45455..77a397ac7db 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -19,6 +19,7 @@ import { heartbeatRunEvents, heartbeatRuns, issueRelations, + issueComments, issues, projects, } from "@paperclipai/db"; @@ -493,7 +494,7 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { return { companyId, agentId, issueId, runId, now }; } - it.each(["remote_workspace_prepare_timeout", "reusable_sandbox_resume_failed", "remote_workspace_restore_failed"].flatMap(errorCode => + it.each(["remote_workspace_prepare_timeout", "reusable_sandbox_resume_failed", "remote_workspace_restore_failed", "reviewed_chat_execution_binding_not_ready"].flatMap(errorCode => (["resume", "paused", "Stop", "dependency"] as const).map(action => [errorCode, action] as const)))("bounds %s bootstrap retries across restart and honors %s", async (errorCode, action) => { const { companyId, agentId, issueId, runId, now } = await seedMaxTurnFixture(); await db.update(agents).set({ adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model: "gpt-6.1-sol" } }) @@ -535,6 +536,39 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(3); }); + it("retains the reviewed chat admission on a durable retry without approving the task", async () => { + const { companyId, agentId, issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" }); + const wakeCommentIds = [randomUUID(), randomUUID()]; + const source = "chat:slack"; + const contextSnapshot = { issueId, taskId: issueId, source, wakeCommentIds, commentId: wakeCommentIds[0] }; + await db.update(heartbeatRuns).set({ + errorCode: "reviewed_chat_execution_binding_not_ready", contextSnapshot, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, runId)); + const outcomes = await Promise.all([ + heartbeat.scheduleBoundedRetry(runId, { now, random: () => 0 }), + heartbeatService(db).scheduleBoundedRetry(runId, { now, random: () => 0 }), + ]); + expect(outcomes.every(outcome => outcome.outcome === "scheduled")).toBe(true); + const children = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId)); + expect(children).toHaveLength(1); + expect(children[0]).toMatchObject({ companyId, agentId, issueId, scopeKind: "issue", status: "scheduled_retry", + scheduledRetryAttempt: 1, scheduledRetryReason: "transient_failure", contextSnapshot }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]) + .toMatchObject({ status: "in_review", executionRunId: children[0]!.id }); + const [laterComment] = await db.insert(issueComments).values({ companyId, issueId, + authorType: "user", authorUserId: "responsible-user", body: "A later board follow-up" }).returning(); + await heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "system", reason: "issue_commented", + payload: { issueId, commentId: laterComment!.id }, requestedByActorType: "user", requestedByActorId: "responsible-user", + contextSnapshot: { issueId, taskId: issueId, source: "issue.comment", wakeCommentId: laterComment!.id, wakeCommentIds: [laterComment!.id] } }); + const [preserved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, children[0]!.id)); + expect(preserved).toEqual(children[0]); + const laterWakes = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"))); + expect(laterWakes).toHaveLength(1); + expect(laterWakes[0]).toMatchObject({ runId: null, payload: { issueId, commentId: laterComment!.id } }); + }); + it("bounds interrupted conversations across restarts and concurrent scheduling", async () => { const { companyId, issueId, runId, now } = await seedMaxTurnFixture(); const resultJson = { conversationContinuation: "continue_conversation_v1" }; diff --git a/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts b/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts index be5b2e57a46..b385ced0352 100644 --- a/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts +++ b/server/src/__tests__/heartbeat-reviewed-chat-binding.integration.test.ts @@ -278,6 +278,43 @@ describe.each(["slack", "discord"] as const)("reviewed %s execution binding", (p await temporary?.cleanup(); }); + it("names exhausted lock contention without accepting authority, then rechecks access after release", async () => { + let release!: () => void; + let acquired!: () => void; + const held = new Promise((resolve) => { acquired = resolve; }); + const gate = new Promise((resolve) => { release = resolve; }); + const lock = db.transaction(async (tx) => { + await tx.select().from(issues).where(eq(issues.id, issueId)).for("update"); + acquired(); + await gate; + }); + await held; + const onDenied = vi.fn(); + try { + await expect(attestReviewedExternalChatRun({ + db, ...binding, contextSnapshot: context, onDenied, + })).rejects.toMatchObject({ + message: "reviewed_chat_execution_binding_not_ready", + code: "reviewed_chat_execution_binding_not_ready", + }); + expect(onDenied).not.toHaveBeenCalled(); + } finally { + release(); + await lock; + } + await expect(attest()).resolves.toBe(true); + await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.principalId, userId)); + try { + await expect(attest()).resolves.toBe(false); + } finally { + await db.update(companyMemberships).set({ status: "active" }).where(eq(companyMemberships.principalId, userId)); + } + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]) + .toMatchObject({ status: "in_review", executionRunId: runId, checkoutRunId: null }); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interactionId)))[0]?.status) + .toBe("pending"); + }, 30_000); + it("attests existing execution ownership while preserving real pending governance and checkout state", async () => { const [before] = await db .select() diff --git a/server/src/modules/wake-queue/domain/context.ts b/server/src/modules/wake-queue/domain/context.ts index ab2d4d5d060..7ce30654bdd 100644 --- a/server/src/modules/wake-queue/domain/context.ts +++ b/server/src/modules/wake-queue/domain/context.ts @@ -83,12 +83,20 @@ export function wakeMustNotJoinActiveRun( ): boolean { // A context compaction run does no task work, so ordinary wakes and // compactions never merge, in either arrival order. - return interactionReceipt || Boolean(parseObject(activeRunContext).interactionId) || + return hasRetainedExternalChatRetry(activeRunContext) || interactionReceipt || Boolean(parseObject(activeRunContext).interactionId) || Boolean(readNonEmptyString(parseObject(activeRunContext).contextCompactionRequestId)) || Boolean(readNonEmptyString(wakeContext.contextCompactionRequestId)) || wakeChatConversationsConflict(activeRunContext, wakeContext); } +export function hasRetainedExternalChatRetry(context: unknown): boolean { + const active = parseObject(context); + const activeSource = readNonEmptyString(active.source); + return readNonEmptyString(active.retryOfRunId) !== null && + (activeSource?.startsWith("chat:") === true || + (activeSource === "issue.interaction.respond" && active.externalChatContinuation === true)); +} + function isInteractionResolutionWakePayload(payload: Record | null | undefined): boolean { return readNonEmptyString(payload?.mutation) === "interaction"; } diff --git a/server/src/modules/wake-queue/index.ts b/server/src/modules/wake-queue/index.ts index b725f6a8052..15e86549d30 100644 --- a/server/src/modules/wake-queue/index.ts +++ b/server/src/modules/wake-queue/index.ts @@ -28,6 +28,7 @@ export type { RunSummary, } from "./application/types.js"; export { WakeQueueApplicationError } from "./application/types.js"; +export { hasRetainedExternalChatRetry } from "./domain/context.js"; export type { IssueSnapshot, RunSnapshot, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 5cdf1ec3a79..4bbbbdb5dc6 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -189,7 +189,7 @@ import type { budgetService } from "../budgets.js"; import { isBoardOwnedBudgetBlock } from "../budget-block-ownership.js"; import type { instanceSettingsService } from "../instance-settings.js"; import type { createRunDispatch } from "../../modules/run-dispatch/index.js"; -import type { PostCommitEffect as WakeQueuePostCommitEffect, createWakeQueue } from "../../modules/wake-queue/index.js"; +import { hasRetainedExternalChatRetry, type PostCommitEffect as WakeQueuePostCommitEffect, type createWakeQueue } from "../../modules/wake-queue/index.js"; type HeartbeatRun = typeof heartbeatRuns.$inferSelect; type HeartbeatRunState = ReturnType; @@ -4566,7 +4566,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // A reviewed chat run attests one conversation, so never fold another // conversation's reply into it; that reply queues its own run instead. const coalescedTargetRun = rawCoalescedTarget && - wakeChatConversationsConflict(rawCoalescedTarget.contextSnapshot, enrichedContextSnapshot) + (hasRetainedExternalChatRetry(rawCoalescedTarget.contextSnapshot) || + wakeChatConversationsConflict(rawCoalescedTarget.contextSnapshot, enrichedContextSnapshot)) ? null : filterZombieCoalesceTarget(rawCoalescedTarget, liveRunExecutions); diff --git a/server/src/services/heartbeat/retries.test.ts b/server/src/services/heartbeat/retries.test.ts index c170776bb97..37ffe770d93 100644 --- a/server/src/services/heartbeat/retries.test.ts +++ b/server/src/services/heartbeat/retries.test.ts @@ -289,6 +289,42 @@ describePostgres("heartbeat retry module database wiring", () => { } }); + it("bounds answered chat attestation retries across restarts and escalates their exhaustion", async () => { + const { run, agent, companyId } = await fixture(); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId: agent.id, + runId: run.id, source: "automation", status: "failed" }).returning(); + const contextSnapshot = { ...run.contextSnapshot, source: "issue.interaction.respond", externalChatContinuation: true, + interactionId: randomUUID(), interactionStatus: "answered", interactionKind: "ask_user_questions", + mutation: "interaction", wakeReason: "issue_commented", wakeCommentIds: [randomUUID()] }; + const [failed] = await db.update(heartbeatRuns).set({ wakeupRequestId: wake!.id, + errorCode: "reviewed_chat_execution_binding_not_ready", contextSnapshot, + }).where(eq(heartbeatRuns.id, run.id)).returning(); + let predecessor = failed!; + for (const attempt of [1, 2]) { + const { retries, deps } = boundRetries(); + expect(await retries.scheduleInteractionContinuationInfrastructureRetryIfEligible(predecessor, agent)) + .toMatchObject({ outcome: "scheduled", attempt, maxAttempts: 2 }); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, predecessor.id)); + expect(successors).toHaveLength(1); + const successor = successors[0]!; + expect(successor).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: attempt, + scheduledRetryReason: legacy.INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + contextSnapshot: { ...contextSnapshot, wakeReason: legacy.INTERACTION_CONTINUATION_INFRA_WAKE_REASON } }); + expect(deps.escalatePlanApprovalResumeFailureNeedsAttention).not.toHaveBeenCalled(); + const [next] = await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), + errorCode: "reviewed_chat_execution_binding_not_ready", resultJson: run.resultJson, + }).where(eq(heartbeatRuns.id, successor.id)).returning(); + predecessor = next!; + } + const { retries, deps } = boundRetries(); + expect(await retries.scheduleInteractionContinuationInfrastructureRetryIfEligible(predecessor, agent)) + .toMatchObject({ outcome: "retry_exhausted", maxAttempts: 2 }); + expect(deps.escalatePlanApprovalResumeFailureNeedsAttention).toHaveBeenCalledOnce(); + expect(deps.escalatePlanApprovalResumeFailureNeedsAttention) + .toHaveBeenCalledWith(expect.objectContaining({ attempt: 2, maxAttempts: 2, run: predecessor })); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(3); + }); + it("keeps the accepted interaction owner and its catalog provider deadline", async () => { const { run, agent } = await fixture(); const retryAt = new Date(now.getTime() + 3600_000); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index bee5fc219af..34a08c0afaf 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -28,6 +28,7 @@ import { REUSABLE_SANDBOX_RESUME_FAILED_CODE } from "../reusable-sandbox-resume- import { PLUGIN_WORKER_UNAVAILABLE_CODE } from "../plugin-worker-unavailable-error.js"; import { ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE } from "../environment-provider-unavailable-error.js"; import { REPOSITORY_CATALOG_UNAVAILABLE_CODE } from "../repository-catalog-unavailable-error.js"; +import { REVIEWED_CHAT_BINDING_NOT_READY } from "../reviewed-chat-binding-diagnostics.js"; import { RUNNER_REVISION_MISMATCH_CODE, RUNNER_REVISION_MISMATCH_RETRY, RUNNER_REVISION_WAIT_RETRY_REASON } from "../runner-revision-mismatch.js"; import { CONTROLLER_LEASE_LOST_FAILURE_CODE } from "../legacy-controller-lease.js"; import { randomUUID } from "node:crypto"; @@ -143,7 +144,7 @@ export function isTransientWorkspacePreparationCode(code: string | null | undefi || code === REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE || code === REUSABLE_SANDBOX_RESUME_FAILED_CODE || code === REMOTE_WORKSPACE_RESTORE_FAILED_CODE || code === PLUGIN_WORKER_UNAVAILABLE_CODE || code === RUNNER_REVISION_MISMATCH_CODE || code === ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE - || code === REPOSITORY_CATALOG_UNAVAILABLE_CODE; + || code === REPOSITORY_CATALOG_UNAVAILABLE_CODE || code === REVIEWED_CHAT_BINDING_NOT_READY; } export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS = @@ -359,6 +360,7 @@ function isRetryableInteractionContinuationInfrastructureFailure( run.errorCode === PLUGIN_WORKER_UNAVAILABLE_CODE || run.errorCode === ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE || run.errorCode === REPOSITORY_CATALOG_UNAVAILABLE_CODE || + run.errorCode === REVIEWED_CHAT_BINDING_NOT_READY || run.errorCode === RUNNER_REVISION_MISMATCH_CODE ) { return true; diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index 599f874aea8..278caf7f769 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -16,7 +16,7 @@ import { createHash, randomUUID } from "node:crypto"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { createNativeRunLogCoverageVerifier, formatNativeRunLogEvent } from "../native-runtime/native-run-log-export.js"; import type { Db } from "@paperclipai/db"; -import { activityLog, agents, agentWakeupRequests, budgetReservations, caseHandoffs, cases, companies, completionContracts, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, heartbeatRunEvents, issueComments, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { activityLog, agents, agentWakeupRequests, budgetReservations, caseHandoffs, cases, companies, completionContracts, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, heartbeatRunEvents, issueComments, issueQuestionResponseDeliveries, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -30,6 +30,7 @@ import { RemoteWorkspaceOperationError, RemoteWorkspaceOperationTimeoutError, Re import { EnvironmentRunError, environmentRunOrchestrator } from "../environment-run-orchestrator.js"; import { ReusableSandboxResumeError } from "../reusable-sandbox-resume-error.js"; import { PluginWorkerUnavailableError } from "../plugin-worker-unavailable-error.js"; +import { ReviewedChatBindingDeniedError, ReviewedChatBindingNotReadyError } from "../reviewed-chat-binding-diagnostics.js"; import { environmentTaskErrorCodeData, JsonRpcCallError, PluginEnvironmentTaskError, PluginWorkspaceOperationError, workspaceOperationErrorData } from "@paperclipai/plugin-sdk"; import { RUNNER_REVISION_MISMATCH_RETRY } from "../runner-revision-mismatch.js"; import { workspaceProviderCallFailure } from "../workspace-provider-failure.js"; @@ -46,6 +47,9 @@ import { buildNativeCompletionContract } from "../native-runtime/completion-cont import { readPendingNativeRunStop } from "../native-runtime/native-cancellation-request.js"; import { reserveRunBudget } from "../budget-reservations.js"; import { hasAcknowledgedNativeReassignmentStopIntent } from "../acknowledged-native-stop.js"; +import { questionResponseDeliveryService } from "../question-response-delivery.js"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; +import { heartbeatService } from "../heartbeat.js"; vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() })); @@ -1044,6 +1048,82 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( } else expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); }); + it.each(["fresh", "started", "untrusted", "denied", "retained", "session", "runner"] as const)("bounds only an owned pre-dispatch chat binding wait: %s", async scenario => { + const f = await fixture({ runtimeMode: scenario === "retained" ? "native" : "legacy", + ...(scenario === "session" ? { nativeSessionId: randomUUID() } : {}), + ...(scenario === "runner" ? { runnerInstanceId: randomUUID() } : {}) }); + const error = scenario === "untrusted" ? new Error("reviewed_chat_execution_binding_not_ready") + : scenario === "denied" ? new ReviewedChatBindingDeniedError("principal_denied") + : new ReviewedChatBindingNotReadyError(); + f.deps.scheduleBoundedRetryForRun.mockResolvedValueOnce({ outcome: "scheduled" } as never); + await f.completion.failRunSetup(f.setup(error, scenario === "started")); + const failed = await read(f.run.id); + expect(failed).toMatchObject({ status: "failed", errorCode: scenario === "fresh" ? "reviewed_chat_execution_binding_not_ready" : "setup_failed" }); + if (scenario === "fresh") { + expect(failed.resultJson).toMatchObject({ executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }); + expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent); + } else expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + }); + + it("keeps an answered chat interaction on its own bounded continuation retry", async () => { + const f = await fixture({ contextSnapshot: { source: "issue.interaction.respond", externalChatContinuation: true, + interactionId: "interaction-1", interactionStatus: "answered", interactionKind: "ask_user_questions", + mutation: "interaction", wakeReason: "issue_commented" } }); + await f.completion.failRunSetup(f.setup(new ReviewedChatBindingNotReadyError())); + expect(await read(f.run.id)).toMatchObject({ status: "failed", errorCode: "reviewed_chat_execution_binding_not_ready" }); + expect(f.deps.scheduleInteractionContinuationInfrastructureRetryIfEligible) + .toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + }); + + it.each([false, true])("selects the answer retry from the real delivery producer and heartbeat queue (legacy context=%s)", async legacy => { + const f = await fixture(); + await db.update(agents).set({ adapterType: "paperclip_runner", status: "idle", + runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agent.id)); + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Answer delivery retry", + status: "in_progress", assigneeAgentId: f.agent.id, responsibleUserId: "local-board" }).returning(); + const [questionRun] = await db.insert(heartbeatRuns).values({ companyId: f.run.companyId, agentId: f.agent.id, + issueId: issue.id, nativeIssueId: issue.id, runtimeMode: "native", status: "succeeded", finishedAt: new Date(), + contextSnapshot: { issueId: issue.id, source: "chat:telegram" } }).returning(); + const interactions = issueThreadInteractionService(db); + const interaction = await interactions.create(issue, { kind: "ask_user_questions", sourceRunId: questionRun.id, + continuationPolicy: "wake_assignee", payload: { version: 1, questions: [{ id: "choice", prompt: "Choose", + selectionMode: "single", required: true, options: [{ id: "yes", label: "Yes" }] }] } }, + { agentId: f.agent.id, runId: questionRun.id }); + await interactions.answerQuestions(issue, interaction.id, { answers: [{ questionId: "choice", optionIds: ["yes"] }] }, + { userId: "local-board" }); + const heartbeat = heartbeatService(db); + const answerHeartbeat = legacy ? { ...heartbeat, wakeup: (...[agentId, options]: Parameters) => { + const contextSnapshot = { ...options?.contextSnapshot }; + delete contextSnapshot.mutation; + return heartbeat.wakeup(agentId, { ...options, contextSnapshot }); + } } : heartbeat; + await questionResponseDeliveryService(db, { heartbeat: answerHeartbeat }).deliver(interaction.id); + const [delivery] = await db.select().from(issueQuestionResponseDeliveries) + .where(eq(issueQuestionResponseDeliveries.interactionId, interaction.id)); + expect(delivery).toMatchObject({ status: "fallback_queued", deliveryMode: "wake_fallback" }); + expect(delivery.targetRunId).toBeTruthy(); + const [admitted] = await db.update(heartbeatRuns).set({ status: "running", executionStage: "preparing", startedAt: new Date() }) + .where(eq(heartbeatRuns.id, delivery.targetRunId!)).returning(); + f.deps.scheduleInteractionContinuationInfrastructureRetryIfEligible.mockImplementation(async run => { + await heartbeat.scheduleBoundedRetry(run.id, { now: new Date(), random: () => 0, + retryReason: "interaction_continuation_infra_retry", wakeReason: "interaction_continuation_infra_retry", maxAttempts: 2 }); + return null; + }); + f.deps.scheduleBoundedRetryForRun.mockResolvedValue({ outcome: "not_scheduled", + reason: "Wrong retry owner fixture", issueId: issue.id }); + await f.completion.failRunSetup({ ...f.setup(new ReviewedChatBindingNotReadyError()), run: admitted, runId: admitted.id }); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + expect(f.deps.scheduleInteractionContinuationInfrastructureRetryIfEligible) + .toHaveBeenCalledWith(expect.objectContaining({ id: admitted.id }), expect.anything()); + const children = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, admitted.id)); + expect(children).toHaveLength(1); + expect(children[0]).toMatchObject({ scheduledRetryReason: "interaction_continuation_infra_retry", + contextSnapshot: { interactionId: interaction.id, interactionStatus: "answered", mutation: "interaction" } }); + expect((await db.select().from(issueQuestionResponseDeliveries) + .where(eq(issueQuestionResponseDeliveries.id, delivery.id)))[0]).toEqual(delivery); + }, 30_000); + const runnerMismatch = () => new JsonRpcCallError({ code: -32002, message: "The installed Runner does not match the expected revision", data: environmentTaskErrorCodeData(new PluginEnvironmentTaskError("runner_revision_mismatch")) }); diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 7a8f2007224..fb071a79ce3 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -25,7 +25,7 @@ import { isWorkspaceValidationFailedRun, isWorkspaceValidationFailure, attachPap import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; import { cancellationResultJson, controllerLeaseLossCancellation, readRunCancellation } from "../run-cancellation.js"; import { CONTROLLER_LEASE_LOST_FAILURE_CODE, LegacyControllerLeaseLostError } from "../legacy-controller-lease.js"; -import { reviewedChatBindingDenialResultJson } from "../reviewed-chat-binding-diagnostics.js"; +import { REVIEWED_CHAT_BINDING_NOT_READY, ReviewedChatBindingNotReadyError, reviewedChatBindingDenialResultJson } from "../reviewed-chat-binding-diagnostics.js"; import { accountRunCost } from "../run-cost-accounting.js"; import { isConversation, settleConversationTurn } from "../agent-conversations.js"; import { legacyExecutionNeedsReconciliationWithEvidence } from "../legacy-execution-recovery.js"; @@ -1396,6 +1396,9 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC && run.nativeSessionId === null && run.runnerInstanceId === null ? findEnvironmentProviderUnavailableError(outerErr) : null; const repositoryCatalogUnavailable = !providerDispatchStarted && run.runtimeMode === "legacy" && run.nativeSessionId === null && run.runnerInstanceId === null ? findRepositoryCatalogUnavailableError(outerErr) : null; + const reviewedChatBindingNotReady = !providerDispatchStarted && run.runtimeMode === "legacy" + && run.nativeSessionId === null && run.runnerInstanceId === null && outerErr instanceof ReviewedChatBindingNotReadyError + ? outerErr : null; const runnerRevisionMismatch = !providerDispatchStarted && run.runtimeMode === "legacy" && run.nativeSessionId === null && run.runnerInstanceId === null && isRunnerRevisionMismatch(outerErr); const setupFailureErrorCode = @@ -1406,6 +1409,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC pluginWorkerUnavailableFailure?.code ?? environmentProviderUnavailable?.code ?? repositoryCatalogUnavailable?.code ?? + reviewedChatBindingNotReady?.code ?? (runnerRevisionMismatch ? RUNNER_REVISION_MISMATCH_CODE : null) ?? workspaceGitScanFailure?.code ?? workspaceValidationSetupFailure?.code ?? @@ -1580,7 +1584,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC // carries the interaction and escalates on exhaustion. const interactionOwnsRetry = (livenessRun.errorCode === PLUGIN_WORKER_UNAVAILABLE_CODE || livenessRun.errorCode === ENVIRONMENT_PROVIDER_UNAVAILABLE_CODE - || livenessRun.errorCode === REPOSITORY_CATALOG_UNAVAILABLE_CODE) + || livenessRun.errorCode === REPOSITORY_CATALOG_UNAVAILABLE_CODE + || livenessRun.errorCode === REVIEWED_CHAT_BINDING_NOT_READY) && isResolvedInteractionContinuationWakeContext(livenessRun.contextSnapshot); await (isTransientWorkspacePreparationCode(livenessRun.errorCode) && !interactionOwnsRetry ? livenessRun.errorCode === RUNNER_REVISION_MISMATCH_CODE diff --git a/server/src/services/heartbeat/run-preparation.ts b/server/src/services/heartbeat/run-preparation.ts index 273ab370bd3..ab1338b01da 100644 --- a/server/src/services/heartbeat/run-preparation.ts +++ b/server/src/services/heartbeat/run-preparation.ts @@ -70,7 +70,7 @@ import { } from "@paperclipai/db"; import { HttpError } from "../../errors.js"; import { isAiConnectionIncompatible } from "../ai-connection-runtime.js"; -import type { ReviewedChatBindingDenialReason } from "../reviewed-chat-binding-diagnostics.js"; +import { ReviewedChatBindingNotReadyError, type ReviewedChatBindingDenialReason } from "../reviewed-chat-binding-diagnostics.js"; import { filterResolvedGitHubConnectionsForRun } from "../git-credentials.js"; import type { AdapterRuntimeMcpAccess, @@ -1768,7 +1768,12 @@ function normalizeInteractionContinuationWakeContext( contextSnapshot: Record, payload: Record | null | undefined, ) { - if (isInteractionResolutionWakePayload(payload)) return; + if (isInteractionResolutionWakePayload(payload)) { + if (contextSnapshot.interactionKind === "ask_user_questions") { + contextSnapshot.mutation = "interaction"; + } + return; + } clearInteractionContinuationWakeContext(contextSnapshot); } @@ -2239,7 +2244,7 @@ export async function attestReviewedExternalChatRun(input: { if (attemptNumber < 50) await new Promise((resolve) => setTimeout(resolve, 100)); } - throw new Error("reviewed_chat_execution_binding_not_ready"); + throw new ReviewedChatBindingNotReadyError(); } export async function readHeartbeatWakeComments(db: Db, companyId: string, issueId: string | null, commentIds: string[]) { diff --git a/server/src/services/native-runtime/external-chat-question-response-retry.ts b/server/src/services/native-runtime/external-chat-question-response-retry.ts new file mode 100644 index 00000000000..72123c98ec5 --- /dev/null +++ b/server/src/services/native-runtime/external-chat-question-response-retry.ts @@ -0,0 +1,52 @@ +import { and, eq } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { INTERACTION_CONTINUATION_INFRA_RETRY_REASON } from "../../modules/run-dispatch/index.js"; +import { executionFailureRetryCount } from "../execution-recovery-attempt.js"; +import { REVIEWED_CHAT_BINDING_NOT_READY } from "../reviewed-chat-binding-diagnostics.js"; +import { nativeSha256 } from "./canonical.js"; + +export const EXTERNAL_CHAT_ANSWER_ADMISSION_KEYS = [ + "source", "issueId", "interactionId", "interactionKind", "interactionStatus", + "sourceRunId", "sourceCommentId", "externalChatContinuation", + "wakeCommentId", "wakeCommentIds", "commentId", +] as const; + +type Run = typeof heartbeatRuns.$inferSelect; +type LockMode = "blocking" | "nonblocking" | "read"; +const record = (value: unknown): Record => + value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +export async function resolveQuestionResponseRetryAdmission( + tx: Db, + run: Run, + lockMode: LockMode, +): Promise { + const seen = new Set(); + let current = run; + while (current.retryOfRunId) { + const context = record(current.contextSnapshot); + if (seen.size >= 2 || seen.has(current.id) || current.scopeKind !== "issue" || current.issueId !== context.issueId || + current.issueId !== run.issueId || current.scheduledRetryReason !== INTERACTION_CONTINUATION_INFRA_RETRY_REASON || + context.retryOfRunId !== current.retryOfRunId || context.retryReason !== INTERACTION_CONTINUATION_INFRA_RETRY_REASON || + context.wakeReason !== INTERACTION_CONTINUATION_INFRA_RETRY_REASON || + context.scheduledRetryAttempt !== current.scheduledRetryAttempt || + current.scheduledRetryAttempt === null || current.scheduledRetryAttempt < 1 || current.scheduledRetryAttempt > 2) return null; + seen.add(current.id); + const query = tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, current.retryOfRunId), + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId))); + const [parent] = await (lockMode === "read" ? query : lockMode === "nonblocking" + ? query.for("update", { noWait: true }) : query.for("update")); + if (!parent || seen.has(parent.id)) return null; + const parentContext = record(parent.contextSnapshot); + const recovery = record(record(parent.resultJson).executionRecovery); + if (parent.scopeKind !== "issue" || parent.issueId !== run.issueId || parent.status !== "failed" || !parent.finishedAt || + parent.errorCode !== REVIEWED_CHAT_BINDING_NOT_READY || parent.runtimeMode !== "legacy" || + parent.nativeIssueId !== null || parent.nativeSessionId !== null || parent.runnerInstanceId !== null || + parent.processPid !== null || parent.processGroupId !== null || parent.externalRunId !== null || + recovery.kind !== "bootstrap" || recovery.providerWorkStarted !== false || + executionFailureRetryCount(parent) + 1 !== current.scheduledRetryAttempt || + EXTERNAL_CHAT_ANSWER_ADMISSION_KEYS.some(key => nativeSha256(parentContext[key] ?? null) !== nativeSha256(context[key] ?? null))) return null; + current = parent; + } + return record(current.contextSnapshot).wakeReason === "issue_commented" ? current : null; +} diff --git a/server/src/services/native-runtime/external-chat-question-response.ts b/server/src/services/native-runtime/external-chat-question-response.ts index 23875f181ec..26abed929cf 100644 --- a/server/src/services/native-runtime/external-chat-question-response.ts +++ b/server/src/services/native-runtime/external-chat-question-response.ts @@ -28,6 +28,8 @@ import { import { questionResponseDeliveryValues } from "../question-response-delivery.js"; import { voiceSessionStore, type VoiceTransaction, currentVoiceCredentialFingerprint } from "../voice/voice-session-store.js"; import { nativeSha256 } from "./canonical.js"; +import { EXTERNAL_CHAT_ANSWER_ADMISSION_KEYS, resolveQuestionResponseRetryAdmission } from "./external-chat-question-response-retry.js"; +import { INTERACTION_CONTINUATION_INFRA_RETRY_REASON } from "../../modules/run-dispatch/index.js"; export const EXTERNAL_CHAT_QUESTION_RESPONSE_KEY = "paperclipExternalChatQuestionResponse"; @@ -221,7 +223,7 @@ async function resolveQuestionResponseChain( if ( context.source !== "issue.interaction.respond" || context.issueId !== binding.issueId || - context.wakeReason !== "issue_commented" || + (context.wakeReason !== "issue_commented" && context.wakeReason !== INTERACTION_CONTINUATION_INFRA_RETRY_REASON) || context.interactionKind !== "ask_user_questions" || context.interactionStatus !== "answered" || context.externalChatContinuation !== true || @@ -251,23 +253,14 @@ async function resolveQuestionResponseChain( : runQuery.for("update")); if (!run?.wakeupRequestId) return null; const admitted = record(run.contextSnapshot); - for (const key of [ - "source", - "issueId", - "interactionId", - "interactionKind", - "interactionStatus", - "sourceRunId", - "sourceCommentId", - "externalChatContinuation", - "wakeCommentId", - "wakeCommentIds", - ]) { + for (const key of [...EXTERNAL_CHAT_ANSWER_ADMISSION_KEYS, "wakeReason"]) { if ( nativeSha256(admitted[key] ?? null) !== nativeSha256(context[key] ?? null) ) return null; } + const admission = await resolveQuestionResponseRetryAdmission(tx, run, lockMode); + if (!admission?.wakeupRequestId) return null; const query = tx .select({ @@ -308,10 +301,10 @@ async function resolveQuestionResponseChain( .innerJoin( agentWakeupRequests, and( - eq(agentWakeupRequests.id, run.wakeupRequestId), + eq(agentWakeupRequests.id, admission.wakeupRequestId), eq(agentWakeupRequests.companyId, issueThreadInteractions.companyId), eq(agentWakeupRequests.agentId, binding.agentId), - eq(agentWakeupRequests.runId, binding.runId), + eq(agentWakeupRequests.runId, admission.id), ), ) .where( @@ -420,7 +413,7 @@ async function resolveQuestionResponseChain( !( (delivery.status === "fallback_queued" && delivery.deliveryMode === "wake_fallback" && - delivery.targetRunId === binding.runId) || + delivery.targetRunId === admission.id) || (delivery.status === "delivering" && delivery.targetRunId === null && delivery.deliveryMode === null) diff --git a/server/src/services/native-runtime/external-chat-wait.integration.test.ts b/server/src/services/native-runtime/external-chat-wait.integration.test.ts index 3bb91b68a38..8145c3ca24c 100644 --- a/server/src/services/native-runtime/external-chat-wait.integration.test.ts +++ b/server/src/services/native-runtime/external-chat-wait.integration.test.ts @@ -88,7 +88,7 @@ import { isExternalChatWaitAuthorizationContention, resolveExternalChatResponseWaitAuthorization, } from "./chat-attachment-reuse.js"; -import { attestReviewedExternalChatRun, buildPaperclipWakePayload } from "../heartbeat.js"; +import { attestReviewedExternalChatRun, buildPaperclipWakePayload, heartbeatService } from "../heartbeat.js"; import { voiceCredentialFingerprint } from "../voice/voice-session-store.js"; import { nativeSha256 } from "./canonical.js"; import { questionResponseDeliveryValues } from "../question-response-delivery.js"; @@ -1297,6 +1297,100 @@ describe("native external-chat response wait", () => { }, ); + async function answerAttestationRetry(attempts: number) { + const native = await seedAnsweredChatTurn(); + const [admitted] = await db.insert(heartbeatRuns).values({ companyId: native.companyId, agentId: native.agentId, + scopeKind: "issue", issueId: native.issueId, status: "running", runtimeMode: "legacy", + responsibleUserId: native.userId, wakeupRequestId: native.wakeId, contextSnapshot: native.context }).returning(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, native.runId)); + await db.update(agentWakeupRequests).set({ runId: admitted!.id }).where(eq(agentWakeupRequests.id, native.wakeId)); + await db.update(issueQuestionResponseDeliveries).set({ targetRunId: admitted!.id }).where(eq(issueQuestionResponseDeliveries.id, native.responseDeliveryId)); + const fixture = { ...native, runId: admitted!.id }; + await db.update(issues).set({ responsibleUserId: fixture.userId, executionRunId: fixture.runId }).where(eq(issues.id, fixture.issueId)); + let predecessorId = fixture.runId; + let child: typeof heartbeatRuns.$inferSelect | undefined; + for (let attempt = 1; attempt <= attempts; attempt += 1) { + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), runtimeMode: "legacy", + nativeIssueId: null, nativeSessionId: null, runnerInstanceId: null, scopeKind: "issue", issueId: fixture.issueId, + responsibleUserId: fixture.userId, errorCode: "reviewed_chat_execution_binding_not_ready", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, predecessorId)); + const scheduled = await heartbeatService(db).scheduleBoundedRetry(predecessorId, { + now: new Date(), random: () => 0, retryReason: "interaction_continuation_infra_retry", + wakeReason: "interaction_continuation_infra_retry", maxAttempts: 2, + }); + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") throw new Error("expected the admitted answer retry"); + child = scheduled.run; + predecessorId = child.id; + } + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, child!.id)); + return { fixture, child: child!, binding: { ...fixture, runId: child!.id }, context: structuredClone(child!.contextSnapshot!) }; + } + + it.each([1, 2])("attests answered retry %s against its unchanged original user receipt", async (attempts) => { + const { fixture, child, binding, context } = await answerAttestationRetry(attempts); + const [delivery] = await db.select().from(issueQuestionResponseDeliveries).where(eq(issueQuestionResponseDeliveries.id, fixture.responseDeliveryId)); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, fixture.wakeId)); + expect(await attestReviewedExternalChatRun({ db, ...binding, contextSnapshot: context })).toBe(true); + context.paperclipExternalChatExecutionBound = true; + context.paperclipWake = await buildPaperclipWakePayload({ db, companyId: fixture.companyId, + agentId: fixture.agentId, runId: child.id, contextSnapshot: context }); + await db.update(heartbeatRuns).set({ contextSnapshot: context }).where(eq(heartbeatRuns.id, child.id)); + expect(await resolveExternalChatQuestionResponse(db, binding, context, "read")) + .toMatchObject({ interactionIds: [fixture.interactionId], marker: { responseDeliveryId: fixture.responseDeliveryId } }); + expect(await materializeExternalChatQuestionResponseInput({ db, binding, contextSnapshot: context })) + .toEqual([expect.objectContaining({ interactionId: fixture.interactionId })]); + expect((await db.select().from(issueQuestionResponseDeliveries).where(eq(issueQuestionResponseDeliveries.id, fixture.responseDeliveryId)))[0]).toEqual(delivery); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, fixture.wakeId)))[0]).toEqual(wake); + }); + + it.each(["revoked_identity", "receipt_actor", "delivery_target", "parent_batch", "provider_started", "parent_session", "unrelated_retry", "foreign_parent", "foreign_task"] as const)( + "rejects answered retry authority after %s", async (change) => { + const { fixture, child, binding, context } = await answerAttestationRetry(1); + if (change === "revoked_identity") await db.update(chatIdentityLinks).set({ status: "revoked" }).where(eq(chatIdentityLinks.principalId, fixture.principalId)); + if (change === "receipt_actor") await db.update(agentWakeupRequests).set({ requestedByActorId: "different-user" }).where(eq(agentWakeupRequests.id, fixture.wakeId)); + if (change === "delivery_target") await db.update(issueQuestionResponseDeliveries).set({ targetRunId: child.id }).where(eq(issueQuestionResponseDeliveries.id, fixture.responseDeliveryId)); + if (change === "parent_batch") await db.update(heartbeatRuns).set({ contextSnapshot: { ...fixture.context, wakeCommentIds: [randomUUID()] } }).where(eq(heartbeatRuns.id, fixture.runId)); + if (change === "provider_started") await db.update(heartbeatRuns).set({ resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: true } } }).where(eq(heartbeatRuns.id, fixture.runId)); + if (change === "parent_session") await db.update(heartbeatRuns).set({ nativeSessionId: randomUUID() }).where(eq(heartbeatRuns.id, fixture.runId)); + if (change === "unrelated_retry") await db.update(heartbeatRuns).set({ scheduledRetryReason: "transient_failure" }).where(eq(heartbeatRuns.id, child.id)); + if (change === "foreign_parent" || change === "foreign_task") { + const [company] = change === "foreign_parent" ? await db.insert(companies).values({ name: "Foreign retry", issuePrefix: `F${randomUUID().replaceAll("-", "")}` }).returning() : []; + const [actor] = company ? await db.insert(agents).values({ companyId: company.id, name: "Foreign agent", adapterType: "process" }).returning() : []; + const companyId = company?.id ?? fixture.companyId; + const [other] = await db.insert(issues).values({ companyId, title: "Another task" }).returning(); + const [foreign] = await db.insert(heartbeatRuns).values({ companyId, agentId: actor?.id ?? fixture.agentId, + scopeKind: "issue", issueId: other!.id, finishedAt: new Date(), + status: "failed", errorCode: "reviewed_chat_execution_binding_not_ready", contextSnapshot: fixture.context, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }).returning(); + context.retryOfRunId = foreign!.id; + await db.update(heartbeatRuns).set({ retryOfRunId: foreign!.id, contextSnapshot: context }).where(eq(heartbeatRuns.id, child.id)); + } + expect(await attestReviewedExternalChatRun({ db, ...binding, contextSnapshot: context })).toBe(false); + expect(context.paperclipExternalChatQuestionResponse).toBeUndefined(); + }, + ); + + it("exhausts answered retries and rejects a forged child beyond their authority bound", async () => { + const { fixture, child } = await answerAttestationRetry(2); + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), + errorCode: "reviewed_chat_execution_binding_not_ready", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, child.id)); + expect(await heartbeatService(db).scheduleBoundedRetry(child.id, { retryReason: "interaction_continuation_infra_retry", + wakeReason: "interaction_continuation_infra_retry", maxAttempts: 3 })).toMatchObject({ outcome: "retry_exhausted" }); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: fixture.companyId, agentId: fixture.agentId, + source: "automation", status: "claimed" }).returning(); + const context = { ...child.contextSnapshot, retryOfRunId: child.id, scheduledRetryAttempt: 3, + executionRetryAccounting: { version: 1, failureRetries: 3, maxTurnContinuations: 0 } }; + const [forged] = await db.insert(heartbeatRuns).values({ ...child, id: randomUUID(), status: "running", + wakeupRequestId: wake!.id, retryOfRunId: child.id, scheduledRetryAttempt: 3, contextSnapshot: context }).returning(); + await db.update(agentWakeupRequests).set({ runId: forged!.id }).where(eq(agentWakeupRequests.id, wake!.id)); + await db.update(issues).set({ executionRunId: forged!.id }).where(eq(issues.id, fixture.issueId)); + expect(await attestReviewedExternalChatRun({ db, ...fixture, runId: forged!.id, contextSnapshot: context })).toBe(false); + }); + async function attestAnswer( fixture: | Awaited> diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index 9e45cb41c90..ec75dbf7881 100644 --- a/server/src/services/question-response-delivery.ts +++ b/server/src/services/question-response-delivery.ts @@ -1212,6 +1212,7 @@ export function questionResponseDeliveryService( interactionId: interaction.id, interactionKind: interaction.kind, interactionStatus: interaction.status, + mutation: "interaction", sourceCommentId: continuationSourceCommentId, sourceRunId: interaction.sourceRunId, externalChatContinuation: externalChatBoundary, diff --git a/server/src/services/reviewed-chat-binding-diagnostics.ts b/server/src/services/reviewed-chat-binding-diagnostics.ts index f1a4727e4df..6661a03fc71 100644 --- a/server/src/services/reviewed-chat-binding-diagnostics.ts +++ b/server/src/services/reviewed-chat-binding-diagnostics.ts @@ -1,4 +1,13 @@ export const REVIEWED_CHAT_BINDING_NOT_AUTHORIZED = "reviewed_chat_execution_binding_not_authorized"; +export const REVIEWED_CHAT_BINDING_NOT_READY = "reviewed_chat_execution_binding_not_ready"; + +export class ReviewedChatBindingNotReadyError extends Error { + readonly code = REVIEWED_CHAT_BINDING_NOT_READY; + + constructor() { + super(REVIEWED_CHAT_BINDING_NOT_READY); + } +} /** * The first reviewed-chat attestation check that denied a run. Closed labels