diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 5f521eefa9c..1e2c236ded9 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2609,6 +2609,13 @@ They read only the queued run's runtime mode. Mutation locks retain complete wake and run rows, so edits, ordering, interruption, and saved execution inputs keep their existing write contract. +The task composer offers Steer when the native session has an active provider +turn that accepts steering. If steering is unavailable or unsupported, it offers +an explicit Interrupt action instead. Interrupt records the board user's intent, +requests cancellation, and delivers the saved queue through normal admission after +the previous execution stops. Preparation and cleanup can keep messages queued; +the action does not bypass Stop confirmation or execution gates. + Run retrieval and session state are in `server/src/services/heartbeat/run-state.ts`. It owns bounded run projections, database encoding checks, task session reads and writes, explicit resumes, session compaction, and usage/billing helpers. diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 1319cdc26b7..21cca819bd5 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -21,6 +21,7 @@ import { issueComments, issueThreadInteractions, issueRecoveryActions, + nativeRunFinalizations, issues, runIdentityContexts, } from "@paperclipai/db"; @@ -78,6 +79,7 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }); afterAll(async () => { + await db?.$client.end(); await tempDb?.cleanup(); }); @@ -814,14 +816,12 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { expect(await db.select().from(heartbeatRuns)).toHaveLength(3); }); - it.each(["stale revision", "native run", "different issue"] as const)( + it.each(["stale revision", "different issue"] as const)( "rejects queued interruption for a %s without stopping the run", async (scenario) => { const seeded = await seedQueue(); - if (scenario !== "native run") { - await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, seeded.agentId)); - await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId)); - } + await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, seeded.agentId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId)); const client = app(seeded.companyId); const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); if (scenario === "different issue") { @@ -835,6 +835,54 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }, ); + it.each(["preparing", "native"] as const)( + "accepts an explicit interrupt fallback for a %s runner and preserves queued messages", + async (runtime) => { + const seeded = await seedQueue(); + if (runtime === "preparing") { + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null, + executionStage: "preparing", + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + }).where(eq(heartbeatRuns.id, seeded.runId)); + } else { + await db.update(heartbeatRuns).set({ nativeIssueId: seeded.issueId }) + .where(eq(heartbeatRuns.id, seeded.runId)); + await db.insert(nativeRunFinalizations).values({ companyId: seeded.companyId, + issueId: seeded.issueId, runId: seeded.runId, phase: "observed" }); + } + // Hold the successor in the ordinary queue without launching a provider. + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }) + .where(eq(agents.id, seeded.agentId)); + await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId, + status: "running", contextSnapshot: { issueId: randomUUID() } }); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" }); + const response = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({ + queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision, + }); + expect(response.status, JSON.stringify(response.body)).toBe(200); + const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(waiting.payload?.queuedCommentInterrupt).toMatchObject({ actorId: "queue-owner", targetRunId: seeded.runId }); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, seeded.issueId))).toHaveLength(2); + expect(waiting.status).toBe("deferred_issue_execution"); + // The synthetic preparation owner now confirms that its unwind finished. + // No provider ran; until this receipt exists, cancellation must hold input. + const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + expect(stopped.status).toBe("cancelled"); + await db.update(heartbeatRuns).set({ resultJson: { ...stopped.resultJson, + startupPreparationSettledAt: new Date().toISOString() } }).where(eq(heartbeatRuns.id, seeded.runId)); + await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(wake.status, JSON.stringify(wake)).toBe("coalesced"); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)); + expect(successor.status).toBe("queued"); + expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds); + await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId); + expect(await db.select().from(heartbeatRuns)).toHaveLength(3); + }, + ); + it("does not accept interruption authority from an agent wake payload", async () => { const seeded = await seedQueue(); await db.update(agents).set({ adapterType: "claude_local", diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 7d5cc8835f6..e05372b5b8b 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -17152,13 +17152,10 @@ export function issueRoutes( allowStoppedTarget: true, }); assertQueueMutationTarget({ queue: locked.queue, queueId: req.body.queueId, revision: req.body.revision }); - const freshResponse = locked.queue.entries.some(entry => entry.source?.requiresFreshSession); - // The UI offers Send whenever no turn is running. The issue's run may - // still be waiting to start or to retry: it has no turn to interrupt, - // so save the click and deliver the queue as soon as that run stops. - const previousRunNotRunning = !locked.activeRun || locked.activeRun.status !== "running"; - const stoppedNativeQueue = locked.queue.protocol === "paperclip_runner_v1" && previousRunNotRunning; - if ((locked.queue.protocol !== "legacy" && !freshResponse && !stoppedNativeQueue) || locked.state !== "deferred" || + // Explicit Interrupt is also the fallback when a native turn cannot + // accept steering. Cancellation and ordinary admission still own Stop + // confirmation, cleanup, and delivery; this click only records intent. + if (locked.state !== "deferred" || !locked.queue.entries.length || (locked.activeRun && locked.activeRun.agentId !== locked.wake.agentId)) { throw conflict("This queue does not support interruption"); @@ -17172,7 +17169,7 @@ export function issueRoutes( await tx.update(agentWakeupRequests).set({ payload: { ...readObject(locked.wake.payload), queuedCommentInterrupt: { actorId: actor.actorId, requestedAt: new Date().toISOString(), - ...(freshResponse && locked.queue.protocol === "paperclip_runner_v1" && locked.activeRun?.status === "running" + ...(locked.queue.protocol === "paperclip_runner_v1" && locked.activeRun?.status === "running" ? { targetRunId: locked.activeRun.id } : {}), } }, updatedAt: new Date(), diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index e86297de002..0a9ed4725cc 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -545,9 +545,9 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0]) .toMatchObject({ result: interaction.result, resolvedByUserId: "board" }); }); - it.each(["matching", "wrong_queue", "wrong_actor", "wrong_target", "missing_receipt", "ordinary_response"])("accepts only the exact live native plan Interrupt receipt: %s", async kind => { + it.each(["matching", "wrong_queue", "wrong_actor", "wrong_target", "missing_receipt", "ordinary_response", "comment"])("accepts only the exact live native queue Interrupt receipt: %s", async kind => { const f = await seed(), queueId = randomUUID(); - const [interaction] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + const [interaction] = kind === "comment" ? [] : await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, sourceRunId: f.sourceRunId, createdByAgentId: f.agentId, kind: "request_confirmation", status: "accepted", resolvedByUserId: "board", resolvedAt: new Date("2026-09-11T09:00:00Z"), payload: { version: 1, prompt: "Approve this plan?" }, result: { version: 1, outcome: "accepted" }, @@ -559,15 +559,18 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes } }).where(eq(heartbeatRuns.id, f.sourceRunId)); await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, source: "on_demand", reason: "issue_commented", status: "deferred_issue_execution", - payload: { issueId: f.issueId, mutation: "interaction", interactionId: interaction.id, interactionStatus: "accepted", - _paperclipWakeContext: { forceFreshSession: kind !== "ordinary_response" }, + payload: { issueId: f.issueId, + ...(kind === "comment" ? { commentId: f.commentId } : { + mutation: "interaction", interactionId: interaction!.id, interactionStatus: "accepted" }), + _paperclipWakeContext: kind === "comment" ? { wakeCommentIds: [f.commentId] } + : { forceFreshSession: kind !== "ordinary_response" }, queuedCommentInterrupt: { actorId: "board", requestedAt: "2026-09-11T09:30:00Z", targetRunId: kind === "wrong_target" ? randomUUID() : f.sourceRunId }, }, }); const result = await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, - commentId: null, queuedCommentInterruptId: queueId, dryRun: true })); - if (kind === "matching") expect(result).toMatchObject({ previousRunId: f.sourceRunId }); + commentId: kind === "comment" ? f.commentId : null, queuedCommentInterruptId: queueId, dryRun: true })); + if (["matching", "ordinary_response", "comment"].includes(kind)) expect(result).toMatchObject({ previousRunId: f.sourceRunId }); else expect(result).toBeNull(); }); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index bedd90e3c3c..9f6a96545ef 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -351,7 +351,7 @@ export async function admitExplicitNativeContinuation(input: { } if (!historicalAdapter && agent.adapterType === "paperclip_runner") return null; } - const interruptedByThisClick = response?.source.requiresFreshSession === true && + const interruptedByThisClick = queuedInterrupt && interruptIntent?.targetRunId === run.id && run.resultJson?.operatorInterrupted === true && run.resultJson?.interruptionSource === "issue_comment_interrupt" && run.resultJson?.interruptedIssueId === issueId && run.resultJson?.interruptedByActorType === "user" && run.resultJson?.interruptedByActorId === actorId && @@ -390,7 +390,8 @@ export async function admitExplicitNativeContinuation(input: { !(queuedRequest && run.runtimeMode === "native" && (run.status !== "cancelled" || authorizedAt > stoppedAt)) && !(queuedRequest && cancelledStartup && authorizedAt > run.finishedAt!) && - !(nativeQueuedSend && (run.runtimeMode === "native" || cancelledStartup) && authorizedAt > run.finishedAt!) && + !(nativeQueuedSend && (run.runtimeMode === "native" || cancelledStartup) && + (authorizedAt > run.finishedAt! || interruptedByThisClick)) && !(queuedInterrupt && response?.source.requiresFreshSession && run.runtimeMode === "native")) return null; // A saved native message sent after Stop is a new user direction. Older // input cannot restart a cancelled run merely because it is still queued. diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index fc99f767c47..2888e573599 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -4849,7 +4849,12 @@ describe("TaskChatThread Paperclip Runner queue", () => { expect(occurrenceCount(queuedComment.body)).toBe(1); }); - it.each(["run-1", null])("offers Interrupt for a live legacy queue and Send for a stopped queue (%s)", (targetRunId) => { + it.each([ + { targetRunId: "run-1", protocol: "legacy", steeringDisposition: "unsupported" }, + { targetRunId: null, protocol: "legacy", steeringDisposition: "unsupported" }, + { targetRunId: "run-1", protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" }, + { targetRunId: "run-1", protocol: "paperclip_runner_v1", steeringDisposition: "unsupported" }, + ] as const)("offers Interrupt for a live $protocol queue with $steeringDisposition steering and Send for a stopped queue ($targetRunId)", ({ targetRunId, protocol, steeringDisposition }) => { const onInterruptQueued = vi.fn(async () => {}); renderThread({ comments: [ @@ -4864,8 +4869,8 @@ describe("TaskChatThread Paperclip Runner queue", () => { queuedCommentQueue: { ...queue, targetRunId, - protocol: "legacy", - steeringDisposition: "unsupported", + protocol, + steeringDisposition, }, onEditQueuedComment: async () => {}, onReorderQueuedComments: async () => {}, diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx index 0423904878d..f44f60b77cd 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx @@ -267,6 +267,36 @@ describe("TaskChatQueuedMessages", () => { ).toBe(true); }); + it.each(["temporarily_unavailable", "unsupported"] as const)( + "offers an explicit interrupt fallback when native steering is %s", + async (steeringDisposition) => { + const props = render({ + queue: { ...queue, steeringDisposition }, + onInterrupt: vi.fn().mockResolvedValue(undefined), + }); + const interrupt = container.querySelector( + '[data-testid="task-chat-queued-interrupt-comment-1"]', + ); + expect(interrupt).not.toBeNull(); + expect(interrupt!.disabled).toBe(false); + expect(interrupt!.textContent).toContain("Interrupt"); + await act(async () => { interrupt!.click(); }); + expect(props.onInterrupt).toHaveBeenCalledOnce(); + expect(props.onSteer).not.toHaveBeenCalled(); + expect(container.querySelector('[data-testid="task-chat-queued-messages"]')).toBeNull(); + }, + ); + + it("keeps same-turn steering when the provider is available, even with an interrupt callback", async () => { + const props = render({ onInterrupt: vi.fn().mockResolvedValue(undefined) }); + expect(container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')).toBeNull(); + await act(async () => { + container.querySelector('[data-testid="task-chat-queued-steer-comment-1"]')!.click(); + }); + expect(props.onSteer).toHaveBeenCalledWith("comment-1", "rev-1"); + expect(props.onInterrupt).not.toHaveBeenCalled(); + }); + it("waits for authoritative discard acknowledgement before removing the row", async () => { const acknowledgement = deferred(); render({ onDiscard: vi.fn().mockReturnValue(acknowledgement.promise) }); diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx index a35eff3b80d..bbc39af7902 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx @@ -156,7 +156,8 @@ function SortableQueuedMessage({ {label} - {(queue.targetRunId || queue.state === "deferred") && (!queue.targetRunId || queue.protocol === "legacy" || entry.source?.requiresFreshSession ? ( + {(queue.targetRunId || queue.state === "deferred") && (!queue.targetRunId || queue.protocol === "legacy" || entry.source?.requiresFreshSession || + (queue.steeringDisposition !== "available" && onInterrupt) ? (