diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts index f0352841235..02a1c12ba35 100644 --- a/server/src/__tests__/task-interruption-query.test.ts +++ b/server/src/__tests__/task-interruption-query.test.ts @@ -90,6 +90,53 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => .toMatchObject({ state: "failed", canRetry: true }); }); + it("shows a queued-message Interrupt as handing off, not as a failed attempt", async () => { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), userId = "board-user"; + await db.insert(companies).values({ id: companyId, name: "Interrupted task", issuePrefix: `Q${companyId.slice(0, 7)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Researcher", adapterType: "paperclip_runner" }); + const [issue] = await db.insert(issues).values({ companyId, title: "Creative brief", + status: "todo", assigneeAgentId: agentId }).returning(); + // The message was queued well before the user clicked Interrupt. + const [wake] = await db.insert(agentWakeupRequests).values({ + companyId, agentId, source: "on_demand", status: "deferred_issue_execution", requestedByActorType: "user", + requestedAt: new Date(Date.now() - 60_000), payload: { issueId: issue.id }, + }).returning(); + const interrupt = async (requestedAt: Date) => { + await db.update(agentWakeupRequests).set({ payload: { issueId: issue.id, + queuedCommentInterrupt: { actorId: userId, requestedAt: requestedAt.toISOString() } } }) + .where(eq(agentWakeupRequests.id, wake.id)); + }; + await interrupt(new Date(Date.now() - 20_000)); + // Same resultJson the queued-comment Interrupt route passes to cancelRun. + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issue.id, scopeKind: "issue", + status: "cancelled", runtimeMode: "native", errorCode: "operator_interrupted", + startedAt: new Date(Date.now() - 42_000), finishedAt: new Date(Date.now() - 20_000), + resultJson: { operatorInterrupted: true, interruptionSource: "issue_comment_interrupt", interruptedIssueId: issue.id, + interruptedByActorType: "user", interruptedByActorId: userId, cancelledByActorType: "user", cancelledByUserId: userId, + queuedCommentInterruptQueueId: wake.id } }); + captureTaskInterruptionFailure.mockClear(); + for (const presentation of [(await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id), + await taskInterruptionPresentation(db, issue)]) { + expect(presentation).toMatchObject({ state: "working", canRetry: false, intentId: wake.id }); + } + expect(captureTaskInterruptionFailure).not.toHaveBeenCalled(); + // A newer queued message does not hide the wake the run was stopped for. + await db.insert(agentWakeupRequests).values({ + companyId, agentId, source: "on_demand", status: "deferred_issue_execution", requestedByActorType: "user", + requestedAt: new Date(Date.now() - 30_000), payload: { issueId: issue.id }, + }); + expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ state: "working", canRetry: false, intentId: wake.id }); + // A run stopped for a different queue is not this hand-off. + await db.update(heartbeatRuns).set({ resultJson: { operatorInterrupted: true, cancelledByActorType: "user", + cancelledByUserId: userId, queuedCommentInterruptQueueId: randomUUID() } }).where(eq(heartbeatRuns.id, runId)); + expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ state: "failed", canRetry: true }); + await db.update(heartbeatRuns).set({ resultJson: { operatorInterrupted: true, cancelledByActorType: "user", + cancelledByUserId: userId, queuedCommentInterruptQueueId: wake.id } }).where(eq(heartbeatRuns.id, runId)); + // A hand-off that never starts eventually offers one Retry. + await interrupt(new Date(Date.now() - 10 * 60_000)); + expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ state: "failed", canRetry: true }); + }); + it("decides several pending native resumes in one list read", async () => { const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Resuming list", issuePrefix: `L${companyId.slice(0, 7)}` }); diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index 73c7682f9be..d4f2e1f07c9 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -28,6 +28,9 @@ const resourceWaitRetryReasons = new Set([WORKSPACE_BUSY_RETRY_REASON, A // A due same-run resume dispatches within seconds; past this, offer Retry. const NATIVE_RESUME_DISPATCH_GRACE_MS = 2 * 60_000; +// A queued-message Interrupt waits for the stopped run's cleanup before its +// successor starts. Past this, a hand-off that never started offers Retry. +const QUEUED_INTERRUPT_HANDOFF_GRACE_MS = 5 * 60_000; type ProjectionIssue = Pick; type ProjectionRun = Pick; projectAccessAllowed?: boolean; projectAccessProjectId?: string | null; @@ -121,6 +128,16 @@ export function projectTaskInterruption(issue: ProjectionIssue, input: Projectio // physical containment and pending cleanup keep their execution-side gates. if (!wait && run.runtimeMode === "native" && run.companyId === issue.companyId && run.nativeIssueId === issue.id && isAcknowledgedNativeStop({ ...run, companyId: run.companyId, nativeIssueId: run.nativeIssueId })) return base; + // A queued-message Interrupt stops this run to deliver that exact message. + // The stop is the user's intent and the message starts next, so show the + // hand-off as working, not as a failed attempt. + // Newer queued input can follow it, so match the recorded queue, not the latest wait. + const interruptWait = input.interruptWait ?? wait; + const interruptAt = Date.parse(interruptWait?.queuedInterruptRequestedAt ?? ""); + if (interruptWait && run.status === "cancelled" && run.resultJson?.queuedCommentInterruptQueueId === interruptWait.id + && Number.isFinite(interruptAt) && now - interruptAt < QUEUED_INTERRUPT_HANDOFF_GRACE_MS + && !(retryWait?.sourceRunId === run.id && retryWait.preparationStatus) && system?.state !== "exhausted") + return { ...base, intentId: interruptWait.id, state: "working" }; if (["running", "queued"].includes(run.status)) return { ...base, state: "working" }; const deniedProject = deniedProjectId(issue, run); let projectAccess: TaskInterruptionPresentation["projectAccess"]; @@ -237,6 +254,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i 'projectAccessDenied', ${heartbeatRuns.resultJson}->'projectAccessDenied', 'cancelledByActorType', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'cancelledByActorType'`, 32)}, 'cancelledByUserId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'cancelledByUserId'`, 160)}, + 'queuedCommentInterruptQueueId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'queuedCommentInterruptQueueId'`, 160)}, 'nativeCancellation', jsonb_build_object( 'schema', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'schema'`, 64)}, 'runId', ${boundedJsonString(sql`${heartbeatRuns.resultJson}->'nativeCancellation'->'runId'`, 160)}, @@ -316,6 +334,19 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i })); } } + // A queued-message Interrupt records its queue on the stopped run. Newer + // input can hide that wake from the latest-wait read, so load it by id. + const interruptQueueId = (run: typeof runs[number]) => { + const id = run.status === "cancelled" ? run.resultJson?.queuedCommentInterruptQueueId : null; + return typeof id === "string" && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/.test(id) ? id : null; + }; + const interruptQueueIds = runs.flatMap(run => interruptQueueId(run) ?? []); + const interruptWaits = interruptQueueIds.length ? await db.select({ issueId: wakeIssue, id: agentWakeupRequests.id, + requestedAt: agentWakeupRequests.requestedAt, + queuedInterruptRequestedAt: boundedJsonString(sql`${agentWakeupRequests.payload}#>'{queuedCommentInterrupt,requestedAt}'`, 40) }) + .from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), inArray(agentWakeupRequests.id, interruptQueueIds), + eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.requestedByActorType, "user"))) : []; + const interruptWaitById = new Map(interruptWaits.map(wait => [wait.id, wait])); const runIds = runs.map(run => run.id); const actions = runIds.length ? await db.selectDistinctOn([issueRecoveryActions.sourceIssueId]).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.sourceIssueId, ids), @@ -354,6 +385,11 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i projectIds.forEach((projectId, index) => currentAccess.set(`${agentId}:${projectId}`, decisions[index]?.allowed === true)); } const actionMap = new Map(actions.map(action => [action.sourceIssueId, action])); + const interruptWaitFor = (issueId: string) => { + const run = runMap.get(issueId); + const wait = run ? interruptWaitById.get(interruptQueueId(run) ?? "") : undefined; + return wait?.issueId === issueId ? wait : undefined; + }; const agentMap = new Map(agentRows.map(agent => [agent.id, agent])); const projectMap = new Map(projectRows.map(project => [project.id, project])); const pauses = new Set(pauseRows.map(row => row.issueId)), approvalsSet = new Set(approvalRows.map(row => row.issueId)); @@ -362,6 +398,7 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i const denied = required?.projectIds.filter(projectId => currentAccess.get(`${issue.assigneeAgentId}:${projectId}`) !== true); const presentation = projectTaskInterruption(issue, { run: runMap.get(issue.id), action: actionMap.get(issue.id), wait: waitMap.get(issue.id), retryWait: retryWaitMap.get(issue.id), + interruptWait: interruptWaitFor(issue.id), projectAccessAllowed: denied?.length === 0, projectAccessProjectId: denied?.includes(issue.projectId ?? "") ? issue.projectId : null, workspaceSelectionAvailable: required?.available ?? false, diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index bdaa13c562e..fd18a489f84 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -5556,6 +5556,30 @@ describe("TaskChatThread live transcript", () => { }, ); + it("renders one Stopped row when an interrupted native turn settles without a reply", () => { + const runId = "native-interrupted"; + nativeTranscriptState.transcriptByRun.set(runId, [ + assistantEntry("2026-11-06T15:14:05Z", "INTERRUPTED-PROGRESS-LINE", "progress"), + { kind: "run_terminal", ts: "2026-11-06T15:14:26Z", turnState: "cancelled", runState: "cancelled", disposition: "blocked" }, + ]); + const activeRun = { + id: runId, runtimeMode: "native" as const, status: "running", invocationSource: "issue" as const, + triggerDetail: null, startedAt: "2026-11-06T15:14:05Z", createdAt: "2026-11-06T15:14:05Z", finishedAt: null, + agentId: "agent-1", agentName: "Researcher", adapterType: "paperclip_runner", + }; + renderThread({ comments: [], onAdd: async () => {}, issueStatus: "in_progress", activeRun }); + // Interrupt cancels the turn. A stopped turn posts no reply comment, so the + // settled turn alone owns it; the live tail must not render a second row. + renderThread({ + comments: [], onAdd: async () => {}, issueStatus: "in_progress", + linkedRuns: [{ ...activeRun, runId, status: "cancelled", finishedAt: "2026-11-06T15:14:26Z", + errorCode: "operator_interrupted", resultJson: { operatorInterrupted: true, cancelledByActorType: "user" } }], + }); + expect(container.querySelector('[data-testid="task-chat-live-transcript"]')).toBeNull(); + expect(container.textContent?.split("INTERRUPTED-PROGRESS-LINE")).toHaveLength(2); + expect(container.textContent?.match(/Stopped (for|after) /g)).toHaveLength(1); + }); + it("keeps the transcript mounted through run settle until the settled turn renders (PAP-462 B4)", () => { transcriptState.transcriptByRun.set("run-1", [ { diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index d4420da495f..bd605379bd9 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -2405,9 +2405,12 @@ export function TaskChatThread(props: TaskChatThreadProps) { // Hand off once the settled turn or its reply comment is in the thread; a // stopped run that yields neither is released by the backstop timeout so the // tail never lingers indefinitely. + const settlingRunSource = + settlingRun != null ? runs.find((run) => run.id === settlingRun.id) : undefined; const settlingIsPaperclipRunner = - settlingRun != null && - isNativePaperclipRunnerRun(runs.find((run) => run.id === settlingRun.id)); + settlingRun != null && isNativePaperclipRunnerRun(settlingRunSource); + // A stopped turn posts no reply, so its settled turn is already complete. + const settlingRunStopped = settlingRunSource?.status === "cancelled"; const settlingHasComment = settlingRun != null && comments.some((comment) => @@ -2434,6 +2437,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { ? settledRunIds.has(settlingRun.id) && (settlingHasComment || settlingHasDurableInteraction || + settlingRunStopped || // A projected terminal answer can already own the response while // the canonical comment request is still in flight. Do not show a // second activity tail just because that fetch is slower.