diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index fe0d043aefa..a23dff7c4f9 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -1866,4 +1866,123 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { expect(countExecuteCallsForRun(runId)).toBe(1); expect(claimedIssue).toEqual({ status: "in_progress", executionRunId: runId }); }); + + async function seedAnsweredQuestion(input: { + companyId: string; + issueId: string; + agentId: string; + sourceRunId: string | null; + rejectedConfirmation?: boolean; + }) { + const interactionId = randomUUID(); + const resolved = { + id: interactionId, + companyId: input.companyId, + issueId: input.issueId, + continuationPolicy: "wake_assignee", + createdByAgentId: input.agentId, + sourceRunId: input.sourceRunId, + resolvedByUserId: "responsible-user", + resolvedAt: new Date(), + }; + await db.insert(issueThreadInteractions).values(input.rejectedConfirmation + ? { + ...resolved, + kind: "request_confirmation", + status: "rejected", + payload: { version: 1, prompt: "Approve the plan?" }, + result: { version: 1, outcome: "rejected", reason: "Needs more work" }, + } + : { + ...resolved, + kind: "ask_user_questions", + status: "answered", + payload: { version: 1, questions: [{ id: "first-task-opening", prompt: "What would you like to do?", selectionMode: "single", + required: true, options: [{ id: "interview", label: "Interview me" }] }] }, + result: { version: 1, answers: [{ questionId: "first-task-opening", optionIds: ["interview"] }] }, + }); + // Mirrors the context question-response delivery queues for an answer. + return { + interactionId, + interactionKind: input.rejectedConfirmation ? "request_confirmation" : "ask_user_questions", + interactionStatus: input.rejectedConfirmation ? "rejected" : "answered", + mutation: "interaction", + sourceRunId: input.sourceRunId, + externalChatContinuation: false, + source: "issue.interaction.respond", + }; + } + + it("starts the assignee when a question no run asked is answered on a todo task that has never started", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(); + let answeredRunId: string | undefined; + let claimedIssue: { status: string; executionRunId: string | null } | null = null; + afterContinuationDispatchCheck = async ({ runId: checkedRunId, issueId: checkedIssueId }) => { + if (checkedIssueId !== issueId || checkedRunId !== answeredRunId) return; + claimedIssue = await db + .select({ status: issues.status, executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + }; + await db.insert(issues).values({ id: issueId, companyId, title: "Onboarding first task", status: "todo", priority: "medium", assigneeAgentId: agentId }); + const contextExtras = await seedAnsweredQuestion({ companyId, issueId, agentId, sourceRunId: null }); + const { runId } = await seedQueuedRun({ companyId, agentId, issueId, wakeReason: "issue_commented", + invocationSource: "automation", contextExtras }); + answeredRunId = runId; + + await heartbeat.resumeQueuedRuns(); + await vi.waitFor(async () => { + const current = await heartbeat.getRun(runId); + expect(countExecuteCallsForRun(runId), JSON.stringify({ + status: current?.status, errorCode: current?.errorCode, + })).toBe(1); + }, { timeout: 5_000 }); + await heartbeat.waitForRunExecutionDrain(runId); + + expect((await heartbeat.getRun(runId))?.status).toBe("succeeded"); + expect(claimedIssue).toEqual({ status: "in_progress", executionRunId: runId }); + expect((await db.select({ startedAt: issues.startedAt }).from(issues).where(eq(issues.id, issueId)))[0]?.startedAt) + .toBeInstanceOf(Date); + }); + + it.each([ + { name: "a started task parked back in todo", startedAt: new Date("2026-01-01T00:00:00.000Z"), askedByRun: false, rejectedConfirmation: false }, + { name: "a question an agent run asked", startedAt: null, askedByRun: true, rejectedConfirmation: false }, + { name: "a rejected confirmation no run asked", startedAt: null, askedByRun: false, rejectedConfirmation: true }, + ])("keeps cancelling an answer continuation for $name", async ({ startedAt, askedByRun, rejectedConfirmation }) => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Parked or agent-asked", status: "todo", priority: "medium", + assigneeAgentId: agentId, startedAt }); + const [sourceRun] = askedByRun + ? await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "assignment", status: "succeeded", + startedAt: new Date(), finishedAt: new Date(), contextSnapshot: { issueId } }).returning() + : [null]; + const contextExtras = await seedAnsweredQuestion({ companyId, issueId, agentId, sourceRunId: sourceRun?.id ?? null, + rejectedConfirmation }); + const { runId, wakeupRequestId } = await seedQueuedRun({ companyId, agentId, issueId, wakeReason: "issue_commented", + invocationSource: "automation", contextExtras }); + + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + const [run, wakeup, issue] = await Promise.all([ + db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null), + db.select({ status: agentWakeupRequests.status, error: agentWakeupRequests.error }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null), + db.select({ status: issues.status }).from(issues).where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null), + ]); + expect(run).toMatchObject({ status: "cancelled", errorCode: "issue_not_in_progress" }); + expect(wakeup).toMatchObject({ status: "skipped", error: expect.stringContaining("no longer in_progress") }); + expect(issue?.status).toBe("todo"); + expect(countExecuteCallsForRun(runId)).toBe(0); + }); }); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 86aee36ca77..0dda710dadf 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -41,6 +41,7 @@ import { evaluateAgentInvokabilityFromDb } from "../../../services/agent-invokab import { budgetService, budgetServiceInTransaction, recheckAdmittedRunBudgetReadOnly } from "../../../services/budgets.js"; import { lockRunBudgetRecoveryOwnerInTransaction, recheckRunBudgetReservationInTransaction } from "../../../services/budget-reservations.js"; import { isCompletedOnboardingHandoffWake } from "../../../services/chat-completion-delivery.js"; +import { isUnstartedTaskQuestionAnswer } from "../../../services/unstarted-task-question-answer.js"; import { isHeartbeatWakeOnDemandEnabled } from "../../../services/heartbeat-policy.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; @@ -666,6 +667,8 @@ export function createPostgresRunDispatchAdapter( issueExecutionRunId: issue?.executionRunId ?? null, issueCheckoutRunId: issue?.checkoutRunId ?? null, isResolvedInteractionContinuation, + isUnstartedTaskQuestionAnswer: Boolean(isResolvedInteractionContinuation && issue?.status === "todo" && + await isUnstartedTaskQuestionAnswer(dbOrTx, { companyId: input.companyId, issueId, interactionId: context.interactionId })), isConnectionContinuation: (isResolvedInteractionContinuation && context.interactionKind === "connection_intent") || context.source === "connection_tools.refreshed", isInteractionWake, diff --git a/server/src/modules/run-dispatch/domain/policy.test.ts b/server/src/modules/run-dispatch/domain/policy.test.ts index 051c676d2c5..8fc730f76b4 100644 --- a/server/src/modules/run-dispatch/domain/policy.test.ts +++ b/server/src/modules/run-dispatch/domain/policy.test.ts @@ -302,6 +302,15 @@ describe("decideQueuedRunStaleness", () => { } satisfies Partial, errorCode: "issue_not_in_progress", }, + { + name: "issue_not_in_progress (resolved-interaction continuation, unstarted answer on a non-todo task)", + overrides: { + isResolvedInteractionContinuation: true, + isUnstartedTaskQuestionAnswer: true, + issueStatus: "backlog", + } satisfies Partial, + errorCode: "issue_not_in_progress", + }, { name: "issue_not_in_progress (max-turn continuation)", overrides: { @@ -399,6 +408,15 @@ describe("decideQueuedRunStaleness", () => { }, NOW)).toEqual({ stale: false }); }); + it("allows an answer to a question no run asked to start a todo task that has never started", () => { + expect(decideQueuedRunStaleness({ + ...baseStalenessFacts(), + isResolvedInteractionContinuation: true, + isUnstartedTaskQuestionAnswer: true, + issueStatus: "todo", + }, NOW)).toEqual({ stale: false }); + }); + it("does not cancel a parked continuation summary when the classifier says it does not park the executor", () => { const facts: QueuedRunFacts = { ...baseStalenessFacts(), diff --git a/server/src/modules/run-dispatch/domain/policy.ts b/server/src/modules/run-dispatch/domain/policy.ts index 9c26cba3eb1..59d955278cd 100644 --- a/server/src/modules/run-dispatch/domain/policy.ts +++ b/server/src/modules/run-dispatch/domain/policy.ts @@ -145,6 +145,8 @@ export type QueuedRunFacts = { issueCheckoutRunId?: string | null; isResolvedInteractionContinuation: boolean; + /** Verified under the issue lock: an answer to a question no run asked, on a todo task that has never started. */ + isUnstartedTaskQuestionAnswer?: boolean; /** A connection resolution or tool refresh can resume an agent waiting in review. */ isConnectionContinuation?: boolean; isInteractionWake: boolean; @@ -510,9 +512,11 @@ export function decideQueuedRunStaleness( } if (facts.isResolvedInteractionContinuation || facts.isConnectionContinuation) { + // A todo task that has never started has no parked execution to protect. + const startsUnstartedTask = facts.issueStatus === "todo" && facts.isUnstartedTaskQuestionAnswer === true; const earlyStatus = decideIssueStatus({ status: facts.issueStatus, - requiresInProgress: facts.issueStatus !== "in_review", + requiresInProgress: facts.issueStatus !== "in_review" && !startsUnstartedTask, terminalBypass: true, }); if (earlyStatus === "not_in_progress") { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9c85c7515c5..43496f50eb8 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,5 +1,6 @@ import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js"; import { isContextCompactionRun } from "./context-compaction.js"; +import { isUnstartedTaskQuestionAnswer } from "./unstarted-task-question-answer.js"; import { registerDeferredReceiptWork } from "./heartbeat/deferred-receipt-work.js"; import { reconcileTaskChatRetryReceipts } from "./task-retry.js"; import { registerIssueRecoveryWork } from "./recovery/work.js"; @@ -2216,11 +2217,14 @@ export function heartbeatService( // Claim the issue under the same active-status predicate used by the // queued-run staleness gate. This is the final atomic guard before // dispatch: an operator parking the issue after claim but before this - // checkout must not be overwritten by the continuation. + // checkout must not be overwritten by the continuation. A todo task + // that has never started has nothing parked, so it may start here. + const startsUnstartedTask = issueContext.status === "todo" && + await isUnstartedTaskQuestionAnswer(db, { companyId: agent.companyId, issueId, interactionId: context.interactionId }); await issuesSvc.checkout( issueId, agent.id, - [...resolvedInteractionCheckoutExpectedStatuses()], + [...resolvedInteractionCheckoutExpectedStatuses(), ...(startsUnstartedTask ? ["todo"] : [])], run.id, ); context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true; diff --git a/server/src/services/unstarted-task-question-answer.ts b/server/src/services/unstarted-task-question-answer.ts new file mode 100644 index 00000000000..15be9115ab0 --- /dev/null +++ b/server/src/services/unstarted-task-question-answer.ts @@ -0,0 +1,41 @@ +import { and, eq, isNull } from "drizzle-orm"; +import { issueThreadInteractions, issues, type Db } from "@paperclipai/db"; +import { isUuidLike } from "@paperclipai/shared"; + +/** + * A question that no run asked, such as the onboarding opening card, can be + * answered before its task has ever started. Answer continuations normally + * require in_progress or in_review so an operator who parked the task wins. + * A todo task that has never started has no parked execution to protect, so + * its answer may check the task out from todo. + * + * Reads only server-owned rows: the interaction must be an answered question + * on this task with no source run, and the task must still be todo with no + * startedAt. Confirmations and other interaction kinds keep the stricter rule. + */ +export async function isUnstartedTaskQuestionAnswer( + db: Db, + input: { companyId: string; issueId: string; interactionId: unknown }, +): Promise { + const interactionId = typeof input.interactionId === "string" ? input.interactionId : null; + if (!interactionId || !isUuidLike(interactionId)) return false; + const rows = await db + .select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions) + .innerJoin(issues, and( + eq(issues.id, issueThreadInteractions.issueId), + eq(issues.companyId, issueThreadInteractions.companyId), + )) + .where(and( + eq(issueThreadInteractions.id, interactionId), + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + eq(issueThreadInteractions.kind, "ask_user_questions"), + eq(issueThreadInteractions.status, "answered"), + isNull(issueThreadInteractions.sourceRunId), + eq(issues.status, "todo"), + isNull(issues.startedAt), + )) + .limit(1); + return rows.length > 0; +}