diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index f6f3e2149b4..a5c1a817681 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -820,8 +820,9 @@ describeEmbeddedPostgres("issue recovery actions", () => { }, ); - it("stands down while the latest run was cancelled by a board operator", async () => { + it.each(["idle", "paused", "error"] as const)("stands down after a board Stop when the assignee is %s", async (agentStatus) => { const { companyId, coderId, sourceIssueId } = await seedCompany(); + await db.update(agents).set({ status: agentStatus }).where(eq(agents.id, coderId)); await db.insert(heartbeatRuns).values({ id: randomUUID(), companyId, @@ -843,6 +844,9 @@ describeEmbeddedPostgres("issue recovery actions", () => { expect(result.operatorCancelExempted).toBe(1); expect(await db.select().from(issueRecoveryActions)).toHaveLength(0); expect(enqueueWakeup).not.toHaveBeenCalled(); + const [issue] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + expect(issue.status).toBe("in_progress"); + expect(issue.assigneeAgentId).toBe(coderId); }); it("stands down after an operator interrupt cancellation", async () => { diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 3b4c78bc075..c3f7b7331d8 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -4723,6 +4723,50 @@ export function issueThreadInteractionService( return expired; }, + // A native callback is no longer answerable after its owned cancellation. + // Historical questions and answers committed concurrently retain their + // normal lifecycle. The bridge supplies the validated native identity. + expireCancelledNativeQuestion: async (input: { + companyId: string; issueId: string; sourceRunId: string; requestId: string; + idempotencyKey: string; agentId: string; + }) => { + const expired = await db.transaction(async (tx) => { + // Answering locks the task before the card. Use the same order so + // cancellation cannot deadlock a concurrent human response. + const [issue] = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.id, input.issueId), eq(issues.companyId, input.companyId), + )).for("no key update").limit(1); + if (!issue) throw notFound("Issue not found"); + const [row] = await tx.update(issueThreadInteractions).set({ + status: "expired", + result: { version: 1, cancelled: true, cancellationReason: "Native question cancelled", answers: [], summaryMarkdown: null }, + resolvedByRunId: input.sourceRunId, resolvedAt: new Date(), updatedAt: new Date(), + }).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + eq(issueThreadInteractions.sourceRunId, input.sourceRunId), + eq(issueThreadInteractions.idempotencyKey, input.idempotencyKey), + eq(issueThreadInteractions.kind, "ask_user_questions"), + eq(issueThreadInteractions.status, "pending"), + sql`${issueThreadInteractions.payload}->>'runtimeRequestId' = ${input.requestId}`, + sql`${issueThreadInteractions.payload}->'questionSet'->>'schema' = 'paperclip.question_set.v1'`, + )).returning(); + if (!row) return null; + const interaction = hydrateInteraction(row) as AskUserQuestionsInteraction; + await enqueueTerminalIssueInteractionChatPublications(tx as unknown as Db, interaction); + await touchIssue(tx, input.issueId); + await logActivity(tx as unknown as Db, { + companyId: input.companyId, actorType: "system", actorId: "native-runtime", + agentId: input.agentId, runId: input.sourceRunId, action: "issue.thread_interaction_expired", + entityType: "issue", entityId: input.issueId, + details: { interactionId: row.id, interactionKind: row.kind, interactionStatus: "expired", reason: "native_question_cancelled" }, + }); + return interaction; + }); + if (expired) await emitInteractionResolvedTelemetry(db, expired); + return expired; + }, + expireConnectionIntentsForOwnershipChange: async (issue: { id: string; companyId: string }) => { const expired = await db.update(issueThreadInteractions).set({ status: "expired", result: { version: 1, outcome: "expired", reason: "The task assignment changed" }, diff --git a/server/src/services/native-runtime/local-native-question-bridge.ts b/server/src/services/native-runtime/local-native-question-bridge.ts index 3ce2cb7c5fb..cc7389bb315 100644 --- a/server/src/services/native-runtime/local-native-question-bridge.ts +++ b/server/src/services/native-runtime/local-native-question-bridge.ts @@ -46,8 +46,6 @@ export function createLocalNativeQuestionBridge(input: { await flushNativeQuestionResponses(input.db, input.binding.runId); }, async observe(event: PrpEvent) { - const request = event.payload.request as Record | undefined; - if (event.eventType !== "runtime_request.created" || request?.type !== "input") return; await projectNativeRuntimeRequest({ db: input.db, binding: input.binding, event }); }, }; diff --git a/server/src/services/native-runtime/native-question-bridge.test.ts b/server/src/services/native-runtime/native-question-bridge.test.ts index 2e1b71f6287..bc9551c2959 100644 --- a/server/src/services/native-runtime/native-question-bridge.test.ts +++ b/server/src/services/native-runtime/native-question-bridge.test.ts @@ -236,6 +236,88 @@ describeEmbeddedPostgres("native question bridge", () => { } } }; } + async function savedRuntimeQuestion() { + const event = runtimeRequestEvent(); + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1, + eventType: event.eventType, sourceSeq: event.sourceSeq, sourceInstanceId: runnerInstanceId, + sourceEventId: event.sourceEventId, payload: { prpEvent: event } }); + const interaction = await projectNativeRuntimeRequest({ db, binding: binding(), event }); + const cancelled: PrpEvent = { ...event, sourceEventId: "runtime-question-cancelled", sourceSeq: 2, + eventType: "runtime_request.cancelled", payload: { requestId: "request-1", requestKind: "runtime", + requestType: "input", turnId: event.turnId, itemId: event.itemId, action: "cancel", reason: "turn_terminal" } }; + return { interaction: interaction!, cancelled }; + } + + it("expires only the cancelled native callback, idempotently, and rejects a late answer", async () => { + await seed(); + const { interaction, cancelled } = await savedRuntimeQuestion(); + const unrelated = await issueThreadInteractionService(db).create({ id: issueId, companyId }, { + kind: "ask_user_questions", payload: interaction.payload, sourceRunId: runId, + idempotencyKey: "ordinary-historical-question", continuationPolicy: "none", + }, { agentId, runId }, { supersedePendingSiblingInteractions: false }); + const bridge = createLocalNativeQuestionBridge({ db, binding: binding(), resolve: vi.fn() }); + await bridge.observe(cancelled); + await bridge.observe(cancelled); + const [expired] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)); + expect(expired).toMatchObject({ status: "expired", result: { cancelled: true, answers: [] } }); + const [historical] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, unrelated.id)); + expect(historical.status).toBe("pending"); + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(issue.status).toBe("in_progress"); + const activity = await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_expired")); + expect(activity).toHaveLength(1); + await expect(issueThreadInteractionService(db).answerQuestions({ id: issueId, companyId }, interaction.id, + { answers: [{ questionId: "color", optionIds: ["blue"] }] }, { userId: "operator-1" })).rejects.toMatchObject({ status: 409 }); + expect(await db.select().from(issueQuestionResponseDeliveries)).toHaveLength(0); + }); + + it.each(["company", "issue", "agent", "run", "session", "runner", "turn", "item", "request"])("rejects cancellation with foreign %s identity", async (field) => { + await seed(); + const { interaction, cancelled } = await savedRuntimeQuestion(); + const owner = binding(); + if (field === "company") owner.companyId = randomUUID(); + if (field === "issue") owner.issueId = randomUUID(); + if (field === "agent") owner.agentId = randomUUID(); + if (field === "run") cancelled.runId = randomUUID(); + if (field === "session") cancelled.normalizedSessionId = randomUUID(); + if (field === "runner") cancelled.sourceInstanceId = randomUUID(); + if (field === "turn") cancelled.turnId = "foreign-turn"; + if (field === "item") cancelled.itemId = "foreign-item"; + if (field === "request") cancelled.payload.requestId = "foreign-request"; + await expect(projectNativeRuntimeRequest({ db, binding: owner, event: cancelled })).rejects.toThrow(/native_runtime_/); + const [current] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)); + expect(current.status).toBe("pending"); + }); + + it("preserves a human answer that committed before native cancellation", async () => { + await seed(); + const { interaction, cancelled } = await savedRuntimeQuestion(); + await issueThreadInteractionService(db).answerQuestions({ id: issueId, companyId }, interaction.id, + { answers: [{ questionId: "color", optionIds: ["blue"] }] }, { userId: "operator-1" }); + await projectNativeRuntimeRequest({ db, binding: binding(), event: cancelled }); + const [current] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)); + expect(current).toMatchObject({ status: "answered", result: { answers: [{ questionId: "color", optionIds: ["blue"] }] } }); + }); + + it("settles concurrent native cancellation and human answer without replay or a deadlock", async () => { + await seed(); + const { interaction, cancelled } = await savedRuntimeQuestion(); + const [stop, answer] = await Promise.allSettled([ + projectNativeRuntimeRequest({ db, binding: binding(), event: cancelled }), + issueThreadInteractionService(db).answerQuestions({ id: issueId, companyId }, interaction.id, + { answers: [{ questionId: "color", optionIds: ["blue"] }] }, { userId: "operator-1" }), + ]); + expect(stop.status).toBe("fulfilled"); + const [current] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)); + if (answer.status === "fulfilled") { + expect(current.status).toBe("answered"); + expect(current.result).toMatchObject({ answers: [{ questionId: "color", optionIds: ["blue"] }] }); + } else { + expect(answer.reason).toMatchObject({ status: 409 }); + expect(current).toMatchObject({ status: "expired", result: { cancelled: true, answers: [] } }); + } + }); + it("commits and acknowledges an ACP permission, then queues only an admin's exact turn-bound decision", async () => { await seed(); const event = permissionRequestEvent(); diff --git a/server/src/services/native-runtime/native-question-bridge.ts b/server/src/services/native-runtime/native-question-bridge.ts index f6ae8d7ae82..309d3ba1512 100644 --- a/server/src/services/native-runtime/native-question-bridge.ts +++ b/server/src/services/native-runtime/native-question-bridge.ts @@ -1,7 +1,7 @@ import { and, eq, inArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { heartbeatRuns, issueThreadInteractions } from "@paperclipai/db"; +import { heartbeatRuns, heartbeatRunEvents, issueThreadInteractions } from "@paperclipai/db"; import type { AskUserQuestionsInteraction, RespondIssueThreadInteraction, @@ -118,7 +118,7 @@ export async function projectNativeRuntimeRequest(input: { binding: Pick; event: PrpEvent; }): Promise { - if (input.event.eventType !== "runtime_request.created") return null; + if (!["runtime_request.created", "runtime_request.cancelled"].includes(input.event.eventType)) return null; if ( input.event.runId !== input.binding.runId || input.event.normalizedSessionId !== input.binding.normalizedSessionId @@ -126,6 +126,47 @@ export async function projectNativeRuntimeRequest(input: { ) { throw new Error("native_runtime_request_binding_mismatch"); } + if (input.event.eventType === "runtime_request.cancelled") { + const outcome = record(input.event.payload)!; + // Permission cards have their own privileged resolver. A provider-loss + // expiry is a durable handoff, not cancellation of a historical question. + if (outcome.requestKind !== "runtime" || outcome.requestType !== "input") return null; + if (outcome.action !== "cancel" || typeof outcome.requestId !== "string" + || !REQUEST_ID_PATTERN.test(outcome.requestId) || typeof input.event.turnId !== "string" + || outcome.turnId !== input.event.turnId || (outcome.itemId ?? null) !== (input.event.itemId ?? null) + || ["response", "answers", "answer", "replay"].some((key) => Object.hasOwn(outcome, key))) { + throw new Error("native_runtime_cancellation_invalid"); + } + const [run] = await input.db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.binding.runId), eq(heartbeatRuns.companyId, input.binding.companyId), + eq(heartbeatRuns.nativeIssueId, input.binding.issueId), eq(heartbeatRuns.agentId, input.binding.agentId), + eq(heartbeatRuns.nativeSessionId, input.binding.normalizedSessionId), + eq(heartbeatRuns.runtimeMode, "native"), + )).limit(1); + if (!run) throw new Error("native_runtime_cancellation_run_mismatch"); + // Match the committed creation, including its turn and item. A run can + // contain several callbacks; its request ID alone must not retire another. + const [created] = await input.db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, input.binding.companyId), + eq(heartbeatRunEvents.runId, input.binding.runId), + eq(heartbeatRunEvents.eventType, "runtime_request.created"), + eq(heartbeatRunEvents.sourceInstanceId, input.binding.runnerSourceInstanceId), + sql`${heartbeatRunEvents.payload}->'prpEvent'->'payload'->'request'->>'requestId' = ${outcome.requestId}`, + sql`${heartbeatRunEvents.sourceSeq} < ${input.event.sourceSeq}`, + )).limit(1); + const original = record(record(created?.payload)?.prpEvent); + const originalRequest = record(record(original?.payload)?.request); + if (!original || original.normalizedSessionId !== input.event.normalizedSessionId + || original.turnId !== input.event.turnId || (original.itemId ?? null) !== (input.event.itemId ?? null) + || originalRequest?.type !== "input" || originalRequest.requestKind !== "runtime") { + throw new Error("native_runtime_cancellation_creation_mismatch"); + } + return issueThreadInteractionService(input.db).expireCancelledNativeQuestion({ + companyId: input.binding.companyId, issueId: input.binding.issueId, agentId: input.binding.agentId, + sourceRunId: input.binding.runId, requestId: outcome.requestId, + idempotencyKey: `${QUESTION_KEY_PREFIX}${input.binding.runId}:${outcome.requestId}`, + }); + } const request = record(record(input.event.payload)?.request); if ( !request diff --git a/server/src/services/native-runtime/runner-prp-coordinator.ts b/server/src/services/native-runtime/runner-prp-coordinator.ts index b71c9f715a3..6d6e30615c0 100644 --- a/server/src/services/native-runtime/runner-prp-coordinator.ts +++ b/server/src/services/native-runtime/runner-prp-coordinator.ts @@ -282,9 +282,7 @@ export function runnerPrpCoordinator( connectionLeaseTtlMs, onCommittedEvent: async (event) => { await nativeStore.appendEvent(event); - if (event.eventType === "runtime_request.created") { - await projectNativeRuntimeRequest({ db, binding: storeBinding, event }); - } + await projectNativeRuntimeRequest({ db, binding: storeBinding, event }); await nativeStore.reconcileTerminalEvent(event); if (event.eventType === "run.terminal") { const stored = await nativeStore.readCompletedRun(); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index ff2de0c3362..371cf45c04b 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -4493,6 +4493,14 @@ export function recoveryService( } } + // Stop remains authoritative even when cancellation accounting or an + // unrelated operator action has since paused the assignee. Check it + // before an uninvokable agent can be escalated as stranded work. + if (issue.status !== "in_review" && isOperatorCancelledRun(latestRun, agentId)) { + result.operatorCancelExempted += 1; + continue; + } + const agent = await getAgent(agentId); const agentInvokable = agent && agent.companyId === issue.companyId