From cc67d4e1d83396e249e3ac27c3f224d432b5385b Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 2 Oct 2026 22:07:14 -0500 Subject: [PATCH] fix: preserve steering and recover stopped task conversations (#15015) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A task conversation must let a user guide a running agent and resume stopped work. > - The active run owns its input protocol, even when the user changes the next model or effort. > - Queue delivery waits for a provider receipt, which must be able to persist during the request. > - A stopped startup also needs a clear user action that passes normal task admission. > - This pull request fixes steering delivery, makes queue actions immediate, and restores explicit continuation. > - The benefit is a responsive conversation that can recover without losing saved input. ## Linked Issues or Issue Description **What happened?** A queued message could change from Steer to Interrupt while a native run prepared. A steer request could wait on its own database lock and fail to deliver. A stopped startup could then leave the conversation without a working Retry or message continuation. Interrupt also waited for the server and showed a toast. **Expected behavior** The active run keeps its input protocol. Steer delivers input to that run. Steer and Interrupt clear the submitted queue rows and show the input in the conversation immediately. Failed delivery restores the latest queue with an inline error. An eligible stopped run offers Retry, and authenticated user input can start a fresh turn through normal task admission. **Steps to reproduce** 1. Start a task with a native Paperclip Runner. 2. Change the selected model or effort while that run prepares. 3. Queue a message and press Steer. 4. Observe the provider receipt and queue state during the request. 5. Stop a startup before its provider process begins, then try Retry or send a new message. 6. Repeat queued delivery with a legacy runner and press Interrupt. **Paperclip version or commit** Reproduced on the parent of this branch, `59c07ede7`. **Deployment mode** Authenticated private deployment. The fixes also cover local task conversations. Related work: Refs #12834, Refs #13354, Refs #13275. The open refactor in #13160 moves the same queue route; it does not fix the receipt lock or stopped-run continuation addressed here. ## What Changed - Select queue behavior from the active run's immutable dispatch and runtime resolution. - Leave the run row unlocked during provider acknowledgement, then lock and read it before merging the receipt. - Retain queued input if the target run stops during that wait. Keep inline delivery errors visible after empty queue updates. - Permit exact Retry and authenticated continuation after verified native startup cancellation. Preserve pause, approval, budget, ownership, and process-stop gates. - Carry undelivered native queue input into a fresh turn once the old execution is confirmed stopped. - Show Steer and Interrupt input in the conversation and clear submitted composer rows immediately. Restore the latest queue inline on failure. Remove delivery toasts. - Keep optimistic delivery stable across stale polls, empty queues, and paginated history. Preserve classic Interrupt error handling. - Document recovery and optimistic delivery behavior. Add regression tests across server, shared queue projection, and UI boundaries. ## Verification - Red-green regression tests reproduced the queue protocol, receipt lock, stopped-startup continuation, and optimistic delivery failures. - The focused server route, continuation, queue, and runner boundary suites passed during implementation. - The queue-route suite passes with 78 tests. The three complete conversation UI suites pass with 347 tests. - UI typecheck, production build, and `pnpm check:token-gates` pass. - Workspace `pnpm -r typecheck` and `pnpm build` pass. The local monolithic `pnpm test:run` is still running; remote CI verifies the complete suite on the latest commit. - All CI gates pass on `bd9031ad56abfcde13d13a13488c1b9217c2fd3a`, including the full test shards, runner verification, browser E2E, typecheck, release registry, and canary dry run. - Greptile reports 5/5 for that commit. Both review threads are resolved. ## Risks This changes queue display and explicit continuation admission. The UI must restore rejected delivery without losing other-session edits. The server must preserve concurrent provider result updates and must not resume a process whose stop is uncertain. Focused tests cover these boundaries. This change has no database migration. ## Model Used OpenAI Codex, an agent based on GPT-6. The exact runtime model ID and context window are not exposed in this session. Used reasoning, repository tools, code execution, and browser inspection. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/SPEC-implementation.md | 8 +- doc/execution-semantics.md | 24 ++- .../shared/src/types/execution-projection.ts | 2 + .../issue-queued-comments-routes.test.ts | 76 ++++++++ .../adapters/queued-comment-postgres.ts | 4 +- .../application/queued-comment-ports.ts | 2 + server/src/routes/agents.ts | 8 +- server/src/routes/issues.ts | 31 +++- .../src/services/cancelled-native-startup.ts | 13 ++ server/src/services/execution-blocker.ts | 7 +- .../explicit-native-continuation.test.ts | 72 ++++++++ .../services/explicit-native-continuation.ts | 19 +- server/src/services/heartbeat.ts | 8 +- .../issue-queued-comment-queue.test.ts | 16 ++ .../services/issue-queued-comment-queue.ts | 13 +- .../native-runtime/native-session-executor.ts | 5 +- .../ExecutionBlockerNotice.test.tsx | 15 ++ ui/src/components/ExecutionBlockerNotice.tsx | 4 +- ui/src/components/TaskChatThread.test.tsx | 22 +++ ui/src/components/TaskChatThread.tsx | 22 ++- .../task-chat/TaskChatQueuedMessages.test.tsx | 58 +++++- .../task-chat/TaskChatQueuedMessages.tsx | 34 ++-- ui/src/lib/issue-queued-comment-queue.test.ts | 12 ++ ui/src/lib/issue-queued-comment-queue.ts | 9 +- ui/src/pages/IssueDetail.test.tsx | 106 +++++++++-- ui/src/pages/IssueDetail.tsx | 166 +++++++++++++----- 26 files changed, 645 insertions(+), 111 deletions(-) diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index f6bda971fab..821c0486fa1 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1702,12 +1702,18 @@ Confirmed project creation appears as a durable card in the shared task transcri ### User continuation after execution recovery stops -An authenticated user message or an exact failed-run Retry can start a fresh +An authenticated user message, a validated undelivered native message queue, or an exact failed-run Retry can start a fresh native or legacy conversation turn once the prior execution is confirmed stopped. Retain the source history and uncertain action outcomes; do not replay tool calls or reset the failed incident's automatic retry budget. Existing pause, approval, budget, ownership, and dependency gates remain in effect. See `doc/execution-semantics.md` for admission and stop-proof requirements. +The task recovery notice offers Retry for eligible failures and verified native +startup cancellations, with failed attempts explained inline. Preparing native +turns keep the Steer label. Steer and Interrupt immediately move the submitted +messages from the composer queue into the conversation while delivery proceeds. +Provider acknowledgement remains authoritative; failed delivery restores the +latest queue with an inline error. Neither action produces a toast. ### Managed AI authentication diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 05b50745ec8..d1130d31819 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -486,7 +486,7 @@ A terminal native failure can retain a result accepted before checkpoint or clea A new user message can continue a terminal native run whose process fields were cleared before local stop receipts existed. Admission must verify the exact run, runner, workspace, and provider session in the retained suspended state, with no active provider turn, pending tool call, or undelivered output. Missing or mismatched state keeps the hold. A later recorded process launch also keeps the hold until its stop is verified. Normal assignment, decision, controller, environment cleanup, and active-run gates still apply. The message starts one fresh conversation turn; it does not replay the failed run, reset its recovery budget, or certify unknown action outcomes. -The task thread exposes the existing guarded Retry action for failed or timed-out legacy conversation runs. Where the server supports an explicit new attempt after a stopped legacy conversation, the thread must not hide that action solely because the old run still has a recovery-needed projection. Native and process recovery holds, pending decisions, active execution, and other retry gates remain in force. When a gate hides Retry, the thread says the message is preserved instead of promising an unavailable action. This presentation change does not rewrite historical outcomes or certify prior actions. +The task thread exposes the guarded Retry action for failed or timed-out conversation runs and native preparation cancelled before provider startup when the never-started proof is verified. The server projects this eligibility on the recovery notice and rechecks it on Retry. Cleanup quarantine retains its inspection path, and non-conversation reconciliation gates remain enforced. Pending decisions, active execution, pause, budget, dependency, and ownership gates remain in force. A refused Retry reports its reason inline and remains available for another attempt. A valid recovery action must name: @@ -1003,7 +1003,7 @@ For native conversations, an authenticated user message sent after the previous Local recovery records a server-authored stop receipt before it clears a verified absent process identity. A new execution request invalidates that receipt before any process can spawn; recording a new process identity also invalidates it. Missing process IDs without a receipt still block admission. Remote execution continues to require termination receipts for every lease. -If cleanup or another execution gate is still pending, the message stays in its existing queue receipt. Startup and periodic scheduling reconsider up to 50 due receipts per pass, at most once per 30 seconds per receipt, without calling a model or resetting recovery attempts. Cleanup callbacks use the same admission path. The issue lock prevents concurrent workers from delivering an adopted or discarded receipt again. The queued-message area shows the current wait reason. Pauses, approvals, budgets, ownership, and external chat authorization remain enforced. A message sent before the run finished does not grant new post-stop authority. +If cleanup or another execution gate is still pending, the message stays in its existing queue receipt. Startup and periodic scheduling reconsider up to 50 due receipts per pass, at most once per 30 seconds per receipt, without calling a model or resetting recovery attempts. Cleanup callbacks use the same admission path. The issue lock prevents concurrent workers from delivering an adopted or discarded receipt again. The queued-message area shows the current wait reason. Pauses, approvals, budgets, ownership, and external chat authorization remain enforced. For a failed native conversation, saved authenticated messages that have never been delivered can request a fresh turn after verified stop even when submitted before the failure finished. Admission validates and adopts their exact queue receipt, order, and original authors. This does not replay the failed request or undo an operator cancellation. Historical legacy interruption holds for conversational adapters no longer block new messages or Resume. Automatic classification uses the server-owned adapter identity saved atomically at run claim, the saved adapter invocation, or the continuation policy, never the agent’s current adapter settings. Missing historical adapter evidence retains the automatic hold; an explicit user continuation can retire it after proving the predecessor stopped. A terminal row with a live predecessor process, an unreleased environment lease, or failed/pending cleanup still blocks actual admission and Resume; a release timestamp alone does not prove cleanup succeeded. Retry scheduling can happen before cleanup, but grants no execution authority. Recovery folds their obsolete no-replay bookkeeping without changing task ownership, status, or automatically waking old work. The audit trail remains readable. Native integrity and ownership holds, and non-conversational adapter holds, remain enforced. @@ -1045,7 +1045,7 @@ unknown, and they do not gain automatic replay eligibility. Admission validates the persisted comment's author, task, and time against every held predecessor. Retry validates the selected failed run's company, task, and agent and preserves that run's identity through admission and history loading. -Duplicate Retry requests adopt the same successor. An agent-authored comment, an old queued request, or a generic +Duplicate Retry requests adopt the same successor. Only a validated, unconsumed human queue receipt can authorize saved-message continuation after a native failure; arbitrary historical input, an agent-authored comment, or a generic system wake cannot release a hold. The source task keeps its assignee. Process ownership, active controllers, cleanup leases, pause, approval, budget, and normal execution gates still apply. Dependency-blocked interaction mode remains limited @@ -1322,8 +1322,22 @@ execution recovery holds still apply. Unconfirmed cleanup does not start work. The active session advertises steering only when its driver supports it. A transport method that rejects steering does not grant that capability. The -queued-message control remains mounted until the server accepts a steer request, -so a rejected last-row action keeps its message and visible error. +queued-message controller remains mounted across an empty queue, preserving +pending delivery and inline errors. Steer immediately moves the selected message +into the conversation; Interrupt immediately moves all submitted queued messages. +Stale queue polls do not restore those rows while delivery is pending or accepted. +Rejected delivery restores the latest queue and shows an inline error without a +toast. The provider acknowledgement remains the authoritative delivery receipt; +optimistic display never grants permission or records successful delivery. A preparing +Paperclip Runner retains the Steer label while steering is unavailable; the +initial legacy database default and an empty queue snapshot cannot select +Interrupt. Model and effort changes apply to the next turn and do not change +the active turn's steering target. While awaiting steering acknowledgement, the +queue route holds the task and queue locks but leaves the run row available for +PRP event persistence. It locks and rereads the run after acknowledgement before +merging the receipt, preserving concurrent provider updates. +If the run stops during that wait, the request fails with a stale-target error +and retains the queued input for continuation. ### Preserve work across handoff and deliver requested files diff --git a/packages/shared/src/types/execution-projection.ts b/packages/shared/src/types/execution-projection.ts index 60090799e9a..02a92849144 100644 --- a/packages/shared/src/types/execution-projection.ts +++ b/packages/shared/src/types/execution-projection.ts @@ -9,6 +9,8 @@ export interface ExecutionBlocker { runError?: string | null; /** Candidate only: admission must still verify termination and all gates. */ canContinue?: boolean; + /** Explicit Retry candidate; the server rechecks stop proof and execution gates. */ + canRetry?: boolean; savedMessageCount?: number; } diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 96e73d597ce..552cab5c486 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -30,6 +30,7 @@ import { issueService } from "../services/issues.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { initializeRunIdentity, reconcileSteeredIdentity } from "../services/run-identity.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -1449,6 +1450,22 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }); }); + it.each(["paperclip_runner", "codex_local"])("keeps Steer during native preparation even after agent settings change to %s", async adapterType => { + const seeded = await seedQueue(); + const steeringCalls = steerNativeSessionMock.mock.calls.length; + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + }).where(eq(heartbeatRuns.id, seeded.runId)); + await db.update(agents).set({ adapterType }).where(eq(agents.id, seeded.agentId)); + const client = app(seeded.companyId); + const initial = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(initial.body).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" }); + const edited = await request(client).patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "Updated direction" }).expect(200); + expect(edited.body).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" }); + expect(steerNativeSessionMock.mock.calls).toHaveLength(steeringCalls); + }); + it("serializes discard against queued-run claim", async () => { const seeded = await seedQueue(); await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1])); @@ -1579,6 +1596,65 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }); }); + it("allows the runner to persist its acknowledgement while steering, preserving concurrent run updates", async () => { + const seeded = await seedQueue(); + await seedDispatchIdentity(seeded); + const originalProfile = { adapterDispatch: { adapterType: "paperclip_runner" }, + nativeExecutionInput: { provider: { kind: "codex", model: "gpt-5.6-sol", reasoningEffort: "medium" } } }; + await db.update(heartbeatRuns).set({ runnerProfileJson: originalProfile }).where(eq(heartbeatRuns.id, seeded.runId)); + await db.update(issues).set({ assigneeAdapterOverrides: { model: "gpt-6-sol", reasoningEffort: "xhigh" } }) + .where(eq(issues.id, seeded.issueId)); + steerNativeSessionMock.mockImplementationOnce(async () => { + await db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout = '1000ms'`); + await appendHeartbeatRunEvent(tx as unknown as typeof db, { + companyId: seeded.companyId, agentId: seeded.agentId, runId: seeded.runId, + eventType: "item.completed", payload: { kind: "steering_acknowledgement" }, + }); + await tx.update(heartbeatRuns).set({ resultJson: { concurrentProviderReceipt: "preserve" } }) + .where(eq(heartbeatRuns.id, seeded.runId)); + }); + return { turnId: "turn-before-model-change" }; + }); + const initial = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`); + const steered = await request(app(seeded.companyId)) + .post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`) + .send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision }); + expect(steered.status, JSON.stringify(steered.body)).toBe(200); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + expect(run.status).toBe("running"); + expect(run.runnerProfileJson).toEqual(originalProfile); + expect(steerNativeSessionMock).toHaveBeenCalledWith(expect.objectContaining({ runId: seeded.runId, + message: "First queued message", correlationId: seeded.commentIds[0] })); + expect(run.resultJson).toMatchObject({ concurrentProviderReceipt: "preserve", queuedSteeringAcknowledgements: { + [seeded.commentIds[0]]: { status: "acknowledged", turnId: "turn-before-model-change" }, + } }); + expect(steered.body.entries.map((entry: any) => entry.comment.id)).toEqual([seeded.commentIds[1]]); + }); + + it.each(["cancelled", "failed", "succeeded"] as const)("preserves queued input when the run stops during acknowledgement (%s)", async status => { + const seeded = await seedQueue(); + await seedDispatchIdentity(seeded); + steerNativeSessionMock.mockImplementationOnce(async () => { + await db.update(heartbeatRuns).set({ status, finishedAt: new Date(), resultJson: { terminalReceipt: "preserve" } }) + .where(eq(heartbeatRuns.id, seeded.runId)); + return { turnId: "turn-that-stopped" }; + }); + const initial = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`); + const response = await request(app(seeded.companyId)) + .post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`) + .send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision }); + expect(response.status).toBe(409); + expect(response.body.details).toMatchObject({ code: "queued_comment_stale_target" }); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(wake.status).toBe("deferred_issue_execution"); + expect((wake.payload as any)._paperclipWakeContext.wakeCommentIds).toEqual(seeded.commentIds); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + expect(run.resultJson).toEqual({ terminalReceipt: "preserve" }); + const activity = await db.select().from(activityLog).where(eq(activityLog.action, "issue.queued_comment_steered")); + expect(activity).toHaveLength(0); + }); + it("keeps the identity pending after a steering timeout, then reconciles it on a later acknowledgement", async () => { const seeded = await seedQueue(); await seedDispatchIdentity(seeded); diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index 375a99eb633..2588636e4bc 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -31,7 +31,9 @@ function toWakeRow(row: WakeRow): QueuedCommentWakeRow { } function toRunRow(row: RunRow): QueuedCommentRunRow { - return { id: row.id, status: row.status, runtimeMode: row.runtimeMode, contextSnapshot: parseObject(row.contextSnapshot) }; + return { id: row.id, status: row.status, runtimeMode: row.runtimeMode, + runtimeModeResolvedAt: row.runtimeModeResolvedAt, runnerProfileJson: row.runnerProfileJson, + contextSnapshot: parseObject(row.contextSnapshot) }; } export type QueuedCommentQueuePostgresAdapterDeps = { diff --git a/server/src/modules/wake-queue/application/queued-comment-ports.ts b/server/src/modules/wake-queue/application/queued-comment-ports.ts index a30d0c6bc02..d3a4f48f01c 100644 --- a/server/src/modules/wake-queue/application/queued-comment-ports.ts +++ b/server/src/modules/wake-queue/application/queued-comment-ports.ts @@ -61,6 +61,8 @@ export type QueuedCommentRunRow = { id: string; status: string; runtimeMode: string | null; + runtimeModeResolvedAt?: Date | null; + runnerProfileJson?: Record | null; contextSnapshot: Record; }; diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index c3f1b8a422c..1fc61ee9e11 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -13,7 +13,7 @@ import { assertAiConnectionCreateAccess, canInstallSharedAiConnectionForNewAgent import { isAiConnectionCompatible } from "@paperclipai/shared"; import { applyConnectorSkills, resolveConnectorAssignments, annotateConnectorSkills, isConnectorSkill } from "../services/connector-runtime.js"; import { getExecutionBlocker } from "../services/execution-blocker.js"; -import { canContinueCancelledRun } from "../services/run-cancellation.js"; +import { canRetryStoppedRun } from "../services/cancelled-native-startup.js"; import { paperclipRunnerTransitionConfig, normalizeLegacyRunnerProvider, isPaperclipRunnerProvider } from "@paperclipai/adapter-utils"; import { executionProjectionForRun, executionProjectionsForRuns } from "../services/execution-projection.js"; import { selectDashboardRunIds } from "../services/dashboard-run-selection.js"; @@ -6006,14 +6006,14 @@ export function agentRoutes( ) { throw notFound("Failed run not found"); } - if (!["failed", "timed_out"].includes(failedRun.status) && !canContinueCancelledRun(failedRun)) { - throw conflict("Only a failed run or verified unexpected cancellation can start a new attempt."); - } if (failedRun.runtimeMode === "native" && failedRun.errorCode === "native_session_cleanup_quarantined") { throw conflict("The stopped native session requires cleanup and reconciliation before a new attempt.", { code: "native_session_cleanup_quarantined", }); } + if (!(await canRetryStoppedRun(db, failedRun))) { + throw conflict("Only a failed run or a verified eligible cancellation can start a new attempt."); + } const failedContext = asRecord(failedRun.contextSnapshot) ?? {}; const issueId = typeof failedContext.issueId === "string" diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 828cfef4703..8b1543a5385 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -7064,6 +7064,8 @@ export function issueRoutes( queueId: string; targetRunId?: string; allowStoppedTarget?: boolean; + /** Steering must allow PRP event ingestion to update the run before ACK. */ + lockRun?: boolean; }) { const [currentIssue] = await input.tx .select() @@ -7145,8 +7147,8 @@ export function issueRoutes( state === "deferred" ? (input.targetRunId ?? input.issue.executionRunId ?? null) : null; - const activeRun = activeRunId - ? await input.tx + const activeRunQuery = activeRunId + ? input.tx .select() .from(heartbeatRuns) .where( @@ -7156,9 +7158,11 @@ export function issueRoutes( input.allowStoppedTarget ? undefined : eq(heartbeatRuns.status, "running"), ), ) - .for("update") - .limit(1) - .then((rows) => rows[0] ?? null) + .$dynamic() + : null; + const activeRun = activeRunQuery + ? await (input.lockRun === false ? activeRunQuery : activeRunQuery.for("update")) + .limit(1).then(rows => rows[0] ?? null) : null; if (input.targetRunId) { const runContext = readObject(activeRun?.contextSnapshot); @@ -15577,7 +15581,6 @@ export function issueRoutes( eq(heartbeatRuns.agentId, retryWake.agentId), ), ) - .for("update") .limit(1) .then((rows) => rows[0] ?? null) : null; @@ -15615,6 +15618,7 @@ export function issueRoutes( actor, queueId: req.body.queueId, targetRunId: req.body.targetRunId, + lockRun: false, }); if (!locked.activeRun) { throw conflict("The queued message targets a stale run", { @@ -15680,6 +15684,17 @@ export function issueRoutes( ? () => reconcileSteeredIdentity(db, steeringIdentity) : undefined, })); + // PRP ingestion allocates event sequences on this row. Locking it + // while awaiting the provider prevents its durable ACK from arriving. + // Re-read under lock after ACK so concurrent run receipts are retained. + const [acknowledgedRun] = await tx.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.id, locked.activeRun.id)).for("update"); + if (!acknowledgedRun || acknowledgedRun.status !== "running") { + throw conflict("The queued message targets a stopped run. Your message is still queued.", { + code: "queued_comment_stale_target", + }); + } + const acknowledgedResult = readObject(acknowledgedRun.resultJson); if (steeringIdentity) await acceptSteeredIdentity(tx, steeringIdentity); acknowledgedTurnId = acknowledgement.turnId; @@ -15711,9 +15726,9 @@ export function issueRoutes( .update(heartbeatRuns) .set({ resultJson: { - ...runResult, + ...acknowledgedResult, queuedSteeringAcknowledgements: { - ...acknowledgements, + ...readObject(acknowledgedResult.queuedSteeringAcknowledgements), [commentId]: { status: "acknowledged", queueId: req.body.queueId, diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 6be8792dda6..db3f082a88a 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -3,10 +3,23 @@ import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinaliza import { claimedAdapterType } from "./conversation-continuation.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; +import { canContinueCancelledRun } from "./run-cancellation.js"; type Run = typeof heartbeatRuns.$inferSelect; type Coordinator = typeof nativeRunFinalizations.$inferSelect; +/** Candidate for explicit user Retry only, never automatic replay after Stop. */ +export async function canRetryStoppedRun(db: Db, run: Run): Promise { + if (run.errorCode === "native_session_cleanup_quarantined") return false; + if (["failed", "timed_out"].includes(run.status) || canContinueCancelledRun(run)) return true; + if (run.status !== "cancelled" || !run.finishedAt || run.processPid || run.processGroupId || + run.processStartedAt || run.sessionIdAfter) return false; + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id), + )); + return isCancelledNativeStartup(db, run, coordinator); +} + /** Caller holds the coordinator and run locks when using this proof to admit * work. Attempt zero is a durable never-claimed receipt: every native executor * commits its first claim before it can start or attach a provider. */ diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 325b9b9d4f5..2d70f70add9 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -3,6 +3,7 @@ import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } import { z } from "zod"; import { agentWakeupRequests, chatConversations, chatEndpoints, heartbeatRuns, issueComments, issues, issueRecoveryActions, toolConnections, type Db } from "@paperclipai/db"; import { canContinueCancelledRun, readRunCancellation } from "./run-cancellation.js"; +import { canRetryStoppedRun } from "./cancelled-native-startup.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; import { EXECUTION_RECONCILIATION_CAUSES, type ExecutionBlocker } from "@paperclipai/shared"; @@ -83,7 +84,9 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st : run?.error); const eligibleContinuation = Boolean(run && conversation?.assigneeAgentId === run.agentId && !["done", "cancelled"].includes(conversation.status) && canContinueCancelledRun(run)); - const [chatBinding] = eligibleContinuation ? await db.select({ + const canRetry = Boolean(run && conversation?.assigneeAgentId === run.agentId && + !["done", "cancelled"].includes(conversation.status) && await canRetryStoppedRun(db, run)); + const [chatBinding] = eligibleContinuation || canRetry ? await db.select({ state: chatConversations.state, endpointStatus: chatEndpoints.status, connectionStatus: toolConnections.status, connectionEnabled: toolConnections.enabled, }).from(chatConversations).leftJoin(chatEndpoints, and( @@ -94,6 +97,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st eq(chatConversations.companyId, companyId), eq(chatConversations.issueId, issueId), )).limit(1) : []; let nextAction = action.nextAction; + if (canRetry && !eligibleContinuation) nextAction += " Try again or send a new message to continue once the previous execution has stopped."; if (chatBinding) { if (chatBinding.state === "endpoint_removed" || !chatBinding.endpointStatus || chatBinding.endpointStatus === "archived") { nextAction = "This chat connection was removed. Inspect the stopped run and create a new task to continue the work."; @@ -117,6 +121,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st runStatus: run?.status ?? null, runError: runError?.slice(0, 1024) ?? null, canContinue: eligibleContinuation && !chatBinding, + canRetry: canRetry && !chatBinding, savedMessageCount: saved.length, }; } diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 830c3ff2645..b535771c0f0 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -1075,6 +1075,78 @@ const support = await getEmbeddedPostgresTestSupport(); expect(coordinator.attempt).toBe(3); expect(coordinator.failureDetail?.replacementDenied).toBe("explicit_user_continuation"); }); + + it.each(["ready", "live_process", "pause", "delivered", "mixed_authors", "last_delivered", "operator_stop"])( + "automatically delivers an unconsumed native message sent before the failed run stopped, respecting %s", async gate => { + const f = await seed(); + await db.update(issueComments).set({ createdAt: new Date("2026-09-11T09:59:00Z") }).where(eq(issueComments.id, f.commentId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + let queuedIds = [f.commentId]; + let expectedIds = queuedIds; + if (["mixed_authors", "last_delivered"].includes(gate)) { + const secondId = randomUUID(); + await db.insert(issueComments).values({ id: secondId, companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "second-author", body: "Also preserve the existing work.", + createdAt: new Date("2026-09-11T09:59:01Z") }); + queuedIds = [f.commentId, secondId]; + expectedIds = gate === "last_delivered" ? [f.commentId] : queuedIds; + if (gate === "last_delivered") await db.update(heartbeatRuns).set({ + contextSnapshot: { issueId: f.issueId, wakeCommentIds: [secondId] }, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + } + if (gate === "live_process") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (gate === "operator_stop") await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "operator_interrupted" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (gate === "delivered") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, + wakeCommentIds: queuedIds } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (gate === "pause") { + const holdId = randomUUID(); + await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause", status: "active" }); + await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId, issueId: f.issueId, depth: 0, issueTitle: "Deploy", issueStatus: "blocked" }); + } + const queueId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "automation", status: "deferred_issue_execution", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board", updatedAt: new Date(0), + payload: { issueId: f.issueId, commentId: queuedIds.at(-1), executionWait: { reason: "execution_recovery" }, + _paperclipWakeContext: { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentIds: queuedIds, wakeCommentId: queuedIds.at(-1) } }, + }); + const service = heartbeatService(db); + await service.resumeExecutionWaitComments(); + await service.resumeExecutionWaitComments(); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + if (!["ready", "mixed_authors", "last_delivered"].includes(gate)) { + expect(receipt.status).toBe("deferred_issue_execution"); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))).toHaveLength(0); + return; + } + expect(receipt.status).toBe("coalesced"); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)); + expect(successor.contextSnapshot).toMatchObject({ wakeCommentIds: expectedIds, + explicitUserContinuation: { previousRunId: f.sourceRunId, commentId: expectedIds.at(-1) } }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + const [prior] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(prior.status).toBe("failed"); + }); + + it.each(["native", "before_selection"])("offers and admits an exact Retry for native preparation cancelled before provider startup: %s", async phase => { + const f = await seedCancelledStartup(); + if (phase === "before_selection") { + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null, nativeIssueId: null, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { startupCancellation: { beforeNativeSelection: true }, startupPreparationSettledAt: new Date().toISOString() }, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + } + const blocker = await getExecutionBlocker(db, f.companyId, f.issueId); + expect(blocker).toMatchObject({ runId: f.sourceRunId, canRetry: true }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const service = heartbeatService(db); + const successor = await service.wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "retry_failed_run", + failedRunId: f.sourceRunId, requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId } }); + expect(successor).toMatchObject({ status: "queued", retryOfRunId: f.sourceRunId }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + }); it.each(["failed", "succeeded_result", "rejected_result", "still_committing", "controller", "successor", "live_process"])( "requires a fully committed failed result and verified stop for a new turn (%s)", async scenario => { const f = await seed(), contractId = randomUUID(), resultId = randomUUID(); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index eb47212a0dd..4dd475b43ab 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -76,7 +76,7 @@ export async function admitExplicitNativeContinuation(input: { failedRunId?: string | null; /** Server-recorded board intent to send an existing legacy message queue. */ queuedCommentInterruptId?: string; - /** Internal delivery of an unconsumed, user-authored legacy queue entry. */ + /** Internal delivery of an unconsumed, user-authored saved queue entry. */ queuedCommentRequestId?: string; dryRun?: boolean; onBlocked?: (reason: string, message: string) => void; @@ -114,10 +114,14 @@ export async function admitExplicitNativeContinuation(input: { !savedQueue.idempotencyKey?.startsWith("chat-inbound:") && queuedCommentIdsFromWakePayload(savedQueue.payload).includes(commentId)); if (input.queuedCommentRequestId && !queuedRequest) return null; + let partiallyDeliveredQueue = false; if (queuedRequest) { const ids = queuedCommentIdsFromWakePayload(savedQueue!.payload); const undelivered = await undeliveredLegacyUserCommentIds(db, companyId, issueId, agentId, ids); - if (undelivered.length !== ids.length) return null; + // A partly consumed queue still owns its remaining input. Admission and + // adoption both recheck the durable queue and discard already delivered IDs. + if (undelivered.at(-1) !== commentId) return null; + partiallyDeliveredQueue = undelivered.length !== ids.length; } const [comment] = retry || response ? [] : await db.select().from(issueComments).where(and( eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), @@ -165,16 +169,17 @@ export async function admitExplicitNativeContinuation(input: { const unusedAdmission = run.status === "cancelled" && !run.startedAt && run.errorCode === "execution_reconciliation_required" && !run.processPid && !run.processGroupId && !run.nativeSessionId; + if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission) return null; const legacyUserTurn = run.runtimeMode === "legacy" && action.cause === "legacy_execution_requires_reconciliation" && isConversationAdapter(agent.adapterType); if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission && + !(queuedRequest && run.runtimeMode === "native" && + (run.status !== "cancelled" || authorizedAt > run.finishedAt)) && !(queuedInterrupt && response?.source.requiresFreshSession && run.runtimeMode === "native")) return null; // Saved input is a request for a new turn, never permission to undo an // operator Stop or redeliver a message already consumed by this run. - if (retry && run.status === "cancelled" && !canContinueCancelledRun(run)) return blocked("cancelled_by_operator", "Inspect the stopped run and send a new message to continue."); - if (queuedRequest && !queuedInterrupt && ((run.status === "cancelled" && !unusedAdmission && !canContinueCancelledRun(run)) || - run.contextSnapshot?.wakeCommentId === commentId || + if (queuedRequest && !queuedInterrupt && (run.contextSnapshot?.wakeCommentId === commentId || (Array.isArray(run.contextSnapshot?.wakeCommentIds) && run.contextSnapshot.wakeCommentIds.includes(commentId)))) return null; if (legacyUserTurn) { const historicalAdapter = await historicalAdapterType(db, run); @@ -200,6 +205,10 @@ export async function admitExplicitNativeContinuation(input: { return blocked("workspace_repair_required", "Verify safe workspace staging or repair before continuing. Your message is saved."); } const cancelledStartup = await isCancelledNativeStartup(db, run, coordinator); + if (queuedRequest && !queuedInterrupt && run.status === "cancelled" && !unusedAdmission && + !canContinueCancelledRun(run) && !(cancelledStartup && authorizedAt > run.finishedAt!)) return null; + if (retry && run.status === "cancelled" && !canContinueCancelledRun(run) && !cancelledStartup) + return blocked("cancelled_by_operator", "Inspect the stopped run and send a new message to continue."); if (cancelledStartup) cancelledStartupIds.add(run.id); if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn && !cancelledStartup) return null; // A provider failure can finish the normal result/assessment commit path. diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1c1e7c64c27..02162d05c83 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -33,6 +33,7 @@ import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/papercli import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js"; import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js"; import { admitExplicitNativeContinuation, undeliveredLegacyUserCommentIds } from "./explicit-native-continuation.js"; +import { canRetryStoppedRun } from "./cancelled-native-startup.js"; import { connectionIntentService } from "./connection-intents.js"; import { managedAiSessionFingerprintConfig, prepareManagedAiRuntime, assertManagedAiProjectAuth, stripAiAuthBindings, isAiConnectionBusy, AI_AUTH_ENV_KEYS } from "./ai-connection-runtime.js"; import { aiConnectionBindingSchema } from "@paperclipai/shared"; @@ -10413,7 +10414,7 @@ export function heartbeatService( let commentId = deriveCommentId(context, payload); let requestedByActorId = wake.requestedByActorId; const reason = readNonEmptyString(context.wakeReason) ?? wake.reason; - if (stoppedNativeContinuation) { + if (run.runtimeMode === "native") { const ids = await undeliveredLegacyUserCommentIds(db, run.companyId, issueId, run.agentId, queuedCommentIdsFromWakePayload(payload)); if (!ids.length) continue; @@ -10443,6 +10444,7 @@ export function heartbeatService( const admitted = await admitExplicitNativeContinuation({ db, companyId: run.companyId, issueId, agentId: run.agentId, actorType: wake.requestedByActorType, actorId: requestedByActorId, reason, commentId, successorRunId: randomUUID(), dryRun: true, + ...(run.runtimeMode === "native" ? { queuedCommentRequestId: wake.id } : {}), onBlocked: (reason, message) => { wait = { reason, message }; }, }); if (!admitted) { @@ -10460,7 +10462,7 @@ export function heartbeatService( await enqueueWakeup(run.agentId, { source: wake.source as WakeupOptions["source"], triggerDetail: (wake.triggerDetail ?? undefined) as WakeupOptions["triggerDetail"], reason, payload, contextSnapshot: context, requestedByActorType: "user", requestedByActorId, - ...(stoppedNativeContinuation ? { queuedCommentRequestId: wake.id } : {}), + ...(run.runtimeMode === "native" ? { queuedCommentRequestId: wake.id } : {}), idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` }, wake.id); break; } @@ -26901,7 +26903,7 @@ export function heartbeatService( if (opts.requestedByActorType !== "user" || !opts.requestedByActorId || reason !== "retry_failed_run" || source !== "on_demand" || triggerDetail !== "manual" || !failed || failed.companyId !== agent.companyId || failed.agentId !== agentId || - (!["failed", "timed_out"].includes(failed.status) && !canContinueCancelledRun(failed)) || + !(await canRetryStoppedRun(db, failed)) || (failed.nativeIssueId ?? readNonEmptyString(failed.contextSnapshot?.issueId)) !== issueId) { throw conflict("The selected failed run cannot be retried for this task."); } diff --git a/server/src/services/issue-queued-comment-queue.test.ts b/server/src/services/issue-queued-comment-queue.test.ts index 854cf986f76..4659954ed1f 100644 --- a/server/src/services/issue-queued-comment-queue.test.ts +++ b/server/src/services/issue-queued-comment-queue.test.ts @@ -67,6 +67,22 @@ describe("decideQueuedCommentQueueSteering", () => { expect(decision).toEqual({ protocol: "paperclip_runner_v1", kind: "probe", steeringRunId: "run-1" }); }); + + it("keeps a preparing Paperclip Runner on the steering protocol until runtime selection finishes", () => { + expect(decideQueuedCommentQueueSteering({ state: "deferred", queueRunRuntimeMode: null, + activeRun: { id: "preparing-run", runtimeMode: "legacy", runtimeModeResolvedAt: null, + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } } }, + assignedAgentAdapterType: "paperclip_runner", queuedCommentCount: 1, + })).toEqual({ protocol: "paperclip_runner_v1", kind: "temporarily_unavailable" }); + }); + + it("preserves the historical adapter when mutable agent settings change", () => { + expect(decideQueuedCommentQueueSteering({ state: "deferred", queueRunRuntimeMode: null, + activeRun: { id: "legacy-run", runtimeMode: "legacy", runtimeModeResolvedAt: null, + runnerProfileJson: { adapterDispatch: { adapterType: "codex_local" } } }, + assignedAgentAdapterType: "paperclip_runner", queuedCommentCount: 1, + })).toEqual({ protocol: "legacy", kind: "unsupported" }); + }); }); describe("buildQueuedCommentQueueSnapshot entry permissions", () => { diff --git a/server/src/services/issue-queued-comment-queue.ts b/server/src/services/issue-queued-comment-queue.ts index 8a31139d873..7e3b5224686 100644 --- a/server/src/services/issue-queued-comment-queue.ts +++ b/server/src/services/issue-queued-comment-queue.ts @@ -128,11 +128,16 @@ export function decideQueuedCommentQueueSteering(facts: { /** The queued run's own runtime mode. Read only when `state` is `"queued"`. */ queueRunRuntimeMode: string | null; /** The currently running turn, if any. Read only when `state` is `"deferred"`. */ - activeRun: { id: string; runtimeMode: string | null } | null; + activeRun: { id: string; runtimeMode: string | null; runtimeModeResolvedAt?: Date | null; runnerProfileJson?: Record | null } | null; assignedAgentAdapterType: string | null; queuedCommentCount: number; }): QueuedCommentQueueSteeringDecision { - const persistedRuntimeMode = + // The default legacy value is not a selection receipt during preparation. + // Use the run's immutable dispatch choice, never the agent's mutable settings. + const preparingNative = facts.state === "deferred" && facts.activeRun?.runtimeMode === "legacy" && + facts.activeRun.runtimeModeResolvedAt === null && + record(facts.activeRun.runnerProfileJson?.adapterDispatch).adapterType === "paperclip_runner"; + const persistedRuntimeMode = preparingNative ? null : facts.state === "queued" ? facts.queueRunRuntimeMode : facts.state === "deferred" @@ -140,7 +145,7 @@ export function decideQueuedCommentQueueSteering(facts: { : null; const protocol: QueuedCommentQueueProtocol = - persistedRuntimeMode === "native" + preparingNative || persistedRuntimeMode === "native" || (persistedRuntimeMode === null && facts.assignedAgentAdapterType === "paperclip_runner") ? "paperclip_runner_v1" : "legacy"; @@ -150,7 +155,7 @@ export function decideQueuedCommentQueueSteering(facts: { } const steeringRun = facts.state === "deferred" ? facts.activeRun : null; - if (!steeringRun || facts.queuedCommentCount === 0) { + if (!steeringRun || preparingNative || facts.queuedCommentCount === 0) { return { protocol, kind: "temporarily_unavailable" }; } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 11c1e5256e4..148a5ebab3d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -6510,8 +6510,9 @@ export async function steerNativeSession(input: { steeringDeliveries.delete(deliveryKey); }); } - // Do not await the persistence callback here: the route holds the run lock - // until acknowledgement. After a timeout this callback can acquire that lock. + // The route still serializes queue mutations on the task until acknowledgement. + // Reconciliation can acquire that task lock after success or a timeout; never + // join it from the provider's acknowledgement path. if (input.onAcknowledged) void delivery.then(input.onAcknowledged).catch(() => undefined); let timeout: ReturnType | null = null; diff --git a/ui/src/components/ExecutionBlockerNotice.test.tsx b/ui/src/components/ExecutionBlockerNotice.test.tsx index 9627098a37d..3683c76883a 100644 --- a/ui/src/components/ExecutionBlockerNotice.test.tsx +++ b/ui/src/components/ExecutionBlockerNotice.test.tsx @@ -106,6 +106,21 @@ describe("stopped task recovery notice", () => { expect(agentsApi.retryFailedRun).toHaveBeenCalledWith("agent", "failed-run", "company"); expect(onRetried).toHaveBeenCalledOnce(); }); + + it.each(["native_continuation_requires_reconciliation", "uncertain_external_action"])("offers Retry for a server-admitted native failure: %s", async cause => { + await act(async () => root.render( + + )); + const button = container.querySelector("button"); + expect(button?.textContent).toBe("Retry"); + vi.mocked(agentsApi.retryFailedRun).mockResolvedValue({} as never); + await act(async () => button!.click()); + expect(agentsApi.retryFailedRun).toHaveBeenCalledWith("agent", "failed-run", "company"); + }); it("shows a failed Retry in the same container and allows another attempt", async () => { vi.mocked(agentsApi.retryFailedRun).mockRejectedValue(new Error("Environment cleanup is still running.")); await act(async () => container.querySelector("button")!.click()); diff --git a/ui/src/components/ExecutionBlockerNotice.tsx b/ui/src/components/ExecutionBlockerNotice.tsx index d80d11b6a36..175e3d9caca 100644 --- a/ui/src/components/ExecutionBlockerNotice.tsx +++ b/ui/src/components/ExecutionBlockerNotice.tsx @@ -45,8 +45,8 @@ export function ExecutionBlockerNotice({ companyId, issueId, blocker, onRetried Inspect run )} - {!requiresInspection && blocker.agentId && blocker.runId && - ((blocker.cause === "legacy_execution_requires_reconciliation" && failedRun) || blocker.canContinue) && ( + {(!requiresInspection || blocker.canRetry) && blocker.agentId && blocker.runId && + ((blocker.cause === "legacy_execution_requires_reconciliation" && failedRun) || blocker.canContinue || blocker.canRetry) && ( diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index c960d3e9c7f..f4d0f046ab1 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -3675,6 +3675,28 @@ describe("TaskChatThread Paperclip Runner queue", () => { expect(occurrenceCount(queuedComment.body)).toBe(1); }); + it.each(["steer", "interrupt"] as const)("preserves inline %s errors after the optimistic last row clears", async action => { + let rejectDelivery!: (error: Error) => void; + const delivery = new Promise((_, reject) => { rejectDelivery = reject; }); + const actionQueue = { ...queue, protocol: action === "interrupt" ? "legacy" as const : queue.protocol }; + const props = { comments: [queuedComment], onAdd: async () => {}, queuedCommentQueue: actionQueue, + onSteerQueuedComment: () => delivery, onInterruptQueued: () => delivery }; + render(); + await act(async () => { + container.querySelector(`[data-testid="task-chat-queued-${action}-queued-prp-1"]`)!.click(); + }); + render(); + expect(container.querySelector('[data-testid="task-chat-queued-messages"]')).toBeNull(); + await act(async () => { + rejectDelivery(new Error("Connection lost")); + await delivery.catch(() => undefined); + }); + await act(async () => { render(); }); + expect(container.textContent).toContain(action === "steer" + ? "Couldn’t steer. Message is still queued." : "Couldn’t interrupt. Message is still queued."); + expect(occurrenceCount(queuedComment.body)).toBe(1); + }); + it.each(["run-1", null])("keeps legacy queued delivery available with target %s", (targetRunId) => { const onInterruptQueued = vi.fn(async () => {}); render( diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index 45800935a12..1c409845c90 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -5,7 +5,7 @@ import { } from "@/lib/issue-detail-performance"; import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; import { workspaceRestoreMarkerDetail } from "@/lib/workspace-restore-marker"; -import type { ActivityEvent, TaskBrowser } from "@paperclipai/shared"; +import type { ActivityEvent, IssueQueuedCommentQueue, TaskBrowser } from "@paperclipai/shared"; import { useProjectCreatedItems } from "@/hooks/useProjectCreatedItems"; import { skillCreatedItems } from "@/components/task-chat/skill-created-items"; import { requiresExecutionReconciliation } from "@paperclipai/shared"; @@ -633,6 +633,19 @@ export function TaskChatThread(props: TaskChatThreadProps) { queuedCommentQueue && queuedCommentQueue.entries.length > 0 ? queuedCommentQueue : null; + const emptyQueuedCommentQueue = useMemo( + () => ({ + issueId: issueId ?? "", + queueId: null, + state: null, + targetRunId: null, + revision: "empty", + protocol: "paperclip_runner_v1", + steeringDisposition: "temporarily_unavailable", + entries: [], + }), + [issueId], + ); const queuedCommentIds = useMemo( () => new Set( @@ -3092,9 +3105,10 @@ export function TaskChatThread(props: TaskChatThreadProps) { className="relative isolate flex flex-col" data-testid="task-chat-composer-stack" > - {queuedMessageQueue && !composerPause ? ( + {!composerPause ? ( { if (!onReorderQueuedComments) @@ -3110,7 +3124,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { await onSteerQueuedComment(commentId, revision); }} onInterrupt={ - onInterruptQueued && queuedMessageQueue.queueId + onInterruptQueued && queuedMessageQueue?.queueId ? async () => { await onInterruptQueued( queuedMessageQueue.targetRunId, diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx index 8595554a717..ab2828867cc 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx @@ -12,10 +12,12 @@ import { function deferred() { let resolve!: (value: T) => void; - const promise = new Promise((innerResolve) => { + let reject!: (error: Error) => void; + const promise = new Promise((innerResolve, innerReject) => { resolve = innerResolve; + reject = innerReject; }); - return { promise, resolve }; + return { promise, resolve, reject }; } const queue: IssueQueuedCommentQueue = { @@ -138,7 +140,7 @@ describe("TaskChatQueuedMessages", () => { expect(next?.map((entry) => entry.position)).toEqual([0, 1]); }); - it("promotes only the selected steering row immediately", async () => { + it("removes the selected steering row before the provider acknowledges it", async () => { const acknowledgement = deferred(); const props = render({ onSteer: vi.fn().mockReturnValue(acknowledgement.promise), @@ -157,18 +159,63 @@ describe("TaskChatQueuedMessages", () => { '[data-testid="task-chat-queued-message-comment-1"]', ), ).toBeNull(); + expect(container.textContent).toContain("Steering queued message."); expect( container.querySelector( '[data-testid="task-chat-queued-message-comment-2"]', ), ).not.toBeNull(); + await act(async () => { + render({ ...props, queue: { ...queue, entries: [...queue.entries] } }); + }); + expect(container.querySelector('[data-testid="task-chat-queued-message-comment-1"]')).toBeNull(); + await act(async () => { acknowledgement.resolve(); await acknowledgement.promise; }); }); + it("clears the entire interrupt queue immediately and restores it inline on failure", async () => { + let rejectInterrupt!: (error: Error) => void; + const interrupted = new Promise((_, reject) => { rejectInterrupt = reject; }); + const props = render({ queue: { ...queue, protocol: "legacy", steeringDisposition: "unsupported" }, + onInterrupt: vi.fn().mockReturnValue(interrupted) }); + await act(async () => { + container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')!.click(); + }); + expect(props.onInterrupt).toHaveBeenCalledOnce(); + expect(container.querySelector('[data-testid="task-chat-queued-messages"]')).toBeNull(); + await act(async () => { rejectInterrupt(new Error("Connection lost")); await interrupted.catch(() => undefined); }); + expect(container.querySelectorAll('[data-testid^="task-chat-queued-message-"]')).toHaveLength(2); + expect(container.textContent).toContain("Couldn’t interrupt. Message is still queued."); + }); + + it.each(["steer", "interrupt"] as const)("shows a delivery error even when the queue becomes empty while %s is pending", async action => { + const delivery = deferred(); + const props = render({ queue: { ...queue, protocol: action === "interrupt" ? "legacy" : "paperclip_runner_v1" }, + onSteer: vi.fn().mockReturnValue(delivery.promise), onInterrupt: vi.fn().mockReturnValue(delivery.promise) }); + await act(async () => { container.querySelector(`[data-testid="task-chat-queued-${action}-comment-1"]`)!.click(); }); + await act(async () => { render({ ...props, queue: { ...props.queue, revision: "empty-poll", entries: [] } }); }); + await act(async () => { delivery.reject(new Error("Connection lost")); await delivery.promise.catch(() => undefined); }); + const error = action === "steer" ? "Couldn’t steer." : "Couldn’t interrupt."; + expect([...container.querySelectorAll('[role="status"]')].some(status => status.textContent?.includes(error))).toBe(true); + expect(container.querySelector('[data-testid="task-chat-queued-message-comment-1"]')).toBeNull(); + }); + + it("does not restore a message discarded elsewhere while steering was pending", async () => { + let rejectSteer!: (error: Error) => void; + const delivery = new Promise((_, reject) => { rejectSteer = reject; }); + const props = render({ onSteer: vi.fn().mockReturnValue(delivery) }); + await act(async () => { container.querySelector('[data-testid="task-chat-queued-steer-comment-1"]')!.click(); }); + await act(async () => { render({ ...props, queue: { ...queue, revision: "rev-2", entries: [queue.entries[1]] } }); }); + await act(async () => { rejectSteer(new Error("Queue changed")); await delivery.catch(() => undefined); }); + expect(container.querySelector('[data-testid="task-chat-queued-message-comment-1"]')).toBeNull(); + expect(container.querySelector('[data-testid="task-chat-queued-message-comment-2"]')).not.toBeNull(); + expect(container.textContent).toContain("Couldn’t steer."); + }); + it("keeps a row queued when steering fails and announces the retryable state", async () => { render({ onSteer: vi.fn().mockRejectedValue(new Error("steering_timeout")), @@ -344,9 +391,6 @@ describe("TaskChatQueuedMessages", () => { container.querySelector( '[data-testid="task-chat-queued-message-comment-1"]', ), - ).not.toBeNull(); - expect(container.textContent).toContain( - "Queued messages will be sent when the previous run has stopped.", - ); + ).toBeNull(); }); }); diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx index 2df4b95c99b..a365a560677 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx @@ -1,4 +1,4 @@ -import { useEffect, useMemo, useState } from "react"; +import { useEffect, useMemo, useRef, useState } from "react"; import { DndContext, KeyboardSensor, @@ -232,6 +232,9 @@ export function TaskChatQueuedMessages({ onDiscard, }: TaskChatQueuedMessagesProps) { const [entries, setEntries] = useState(queue.entries); + const latestQueue = useRef(queue); + latestQueue.current = queue; + const optimisticDeliveryIds = useRef(new Set()); const [pending, setPending] = useState<{ commentId: string; action: Exclude; @@ -247,7 +250,10 @@ export function TaskChatQueuedMessages({ ); useEffect(() => { - setEntries(queue.entries); + // Polling can return the old queue while delivery is still in flight. + setEntries(queue.entries.filter( + (entry) => !optimisticDeliveryIds.current.has(entry.comment.id), + )); }, [queue.entries, queue.revision]); const ids = useMemo( @@ -309,7 +315,15 @@ export function TaskChatQueuedMessages({ ) { return; } - const previous = entries; + const deliveredIds = action === "interrupt" + ? entries.map((entry) => entry.comment.id) + : action === "steer" ? [commentId] : []; + for (const id of deliveredIds) optimisticDeliveryIds.current.add(id); + if (deliveredIds.length) { + setEntries((current) => current.filter( + (entry) => !optimisticDeliveryIds.current.has(entry.comment.id), + )); + } setPending({ commentId, action }); setVisibleError(null); setAnnouncement( @@ -319,11 +333,6 @@ export function TaskChatQueuedMessages({ ? "Sending queued messages." : "Discarding queued message.", ); - if (action === "steer") { - setEntries((current) => - current.filter((entry) => entry.comment.id !== commentId), - ); - } try { if (action === "steer") await onSteer(commentId, queue.revision); else if (action === "interrupt") await onInterrupt?.(); @@ -341,7 +350,12 @@ export function TaskChatQueuedMessages({ : "Queued message discarded.", ); } catch (error) { - if (action === "steer") setEntries(previous); + for (const id of deliveredIds) optimisticDeliveryIds.current.delete(id); + if (deliveredIds.length) { + setEntries(latestQueue.current.entries.filter( + (entry) => !optimisticDeliveryIds.current.has(entry.comment.id), + )); + } setAnnouncement(""); const code = queueActionErrorCode(error); setVisibleError( @@ -360,7 +374,7 @@ export function TaskChatQueuedMessages({ } } - if (entries.length === 0) return null; + if (entries.length === 0 && !visibleError) return null; return (
{ }); }); + it("keeps native steering while an empty queue snapshot predates the submitted message", () => { + const authoritativeQueue = normalizeIssueQueuedCommentQueue({ + protocol: "legacy", entries: [], + }, "issue-1"); + const queue = mergePendingIssueQueuedComments({ + issueId: "issue-1", authoritativeQueue, + pendingComments: [{ comment: comment("optimistic-1", "Use the new model next turn"), targetRunId: "run-native" }], + fallbackProtocol: "paperclip_runner_v1", + }); + expect(queue).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" }); + }); + it("deduplicates acknowledged entries and restores the authoritative queue identity", () => { const pending = comment("comment-1", "Use three seconds instead"); const authoritativeQueue = normalizeIssueQueuedCommentQueue( diff --git a/ui/src/lib/issue-queued-comment-queue.ts b/ui/src/lib/issue-queued-comment-queue.ts index 8c214f4177c..9cae6860d8a 100644 --- a/ui/src/lib/issue-queued-comment-queue.ts +++ b/ui/src/lib/issue-queued-comment-queue.ts @@ -134,8 +134,11 @@ export function mergePendingIssueQueuedComments(params: { const fallbackTargetRunId = params.pendingComments.find((entry) => entry.targetRunId)?.targetRunId ?? null; - const protocol = - params.authoritativeQueue?.protocol ?? params.fallbackProtocol; + // An empty snapshot predating this submission does not own its runtime. + // Use the active run until the server acknowledges an actual queue. + const protocol = authoritativeOwnsQueue + ? params.authoritativeQueue!.protocol + : params.fallbackProtocol; const targetRunId = authoritativeOwnsQueue ? (params.authoritativeQueue?.targetRunId ?? null) : fallbackTargetRunId; @@ -151,7 +154,7 @@ export function mergePendingIssueQueuedComments(params: { revision: params.authoritativeQueue?.revision ?? "awaiting-server", protocol, steeringDisposition: - params.authoritativeQueue?.steeringDisposition ?? + (authoritativeOwnsQueue ? params.authoritativeQueue?.steeringDisposition : undefined) ?? (protocol === "paperclip_runner_v1" && targetRunId ? "temporarily_unavailable" : "unsupported"), diff --git a/ui/src/pages/IssueDetail.test.tsx b/ui/src/pages/IssueDetail.test.tsx index 057c8570fb6..526f8dc9677 100644 --- a/ui/src/pages/IssueDetail.test.tsx +++ b/ui/src/pages/IssueDetail.test.tsx @@ -755,10 +755,12 @@ function createTooltipRoot(container: Element): Root { function createDeferred() { let resolve!: (value: T) => void; - const promise = new Promise((innerResolve) => { + let reject!: (error: Error) => void; + const promise = new Promise((innerResolve, innerReject) => { resolve = innerResolve; + reject = innerReject; }); - return { promise, resolve }; + return { promise, resolve, reject }; } function createIssue(overrides: Partial = {}): Issue { @@ -3925,11 +3927,11 @@ describe("IssueDetail", () => { expect(mockHeartbeatsApi.cancel).not.toHaveBeenCalled(); }); - it("projects a native follow-up into the steering well before the post resolves", async () => { + it.each(["native", "legacy"] as const)("projects a native follow-up into the steering well before the post resolves with %s persisted mode", async (runtimeMode) => { const postedComment = createDeferred(); const activeRun = { id: "run-native", - runtimeMode: "native" as const, + runtimeMode, status: "running", invocationSource: "issue", triggerDetail: null, @@ -3961,6 +3963,7 @@ describe("IssueDetail", () => { state: null, targetRunId: null, entries: [], + protocol: "legacy", steeringDisposition: "temporarily_unavailable", }), ); @@ -5875,11 +5878,11 @@ describe("IssueDetail", () => { }, ); - it("keeps the last queued message mounted until steering is acknowledged so rejection stays visible", async () => { + it.each([true, false])("optimistically promotes the last steering message and restores it without a toast on rejection (history loaded: %s)", async historyLoaded => { const queue = createQueuedCommentQueue(); mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", executionRunId: "run-active-1" })); mockAgentsApi.list.mockResolvedValue([createAgent({ adapterType: "paperclip_runner" })]); - mockIssuesApi.listComments.mockResolvedValue([queue.entries[0].comment]); + mockIssuesApi.listComments.mockResolvedValue(historyLoaded ? [queue.entries[0].comment] : []); mockIssuesApi.getQueuedComments.mockResolvedValue(queue); mockHeartbeatsApi.activeRunForIssue.mockResolvedValue({ id: "run-active-1", runtimeMode: "native", status: "running", invocationSource: "issue", @@ -5900,13 +5903,18 @@ describe("IssueDetail", () => { }); let pending!: Promise; await act(async () => { pending = props.onSteerQueuedComment("queued-comment-1", queue.revision).catch(error => error); }); - // The child owns its pending/error state. Unmounting it here loses any later error. const whilePending = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props; - expect(whilePending.queuedCommentQueue?.entries).toHaveLength(1); + expect(whilePending.queuedCommentQueue).toBeNull(); + expect(mockIssueChatThreadRender.mock.calls.at(-1)?.[0]?.comments).toContainEqual(expect.objectContaining({ + id: "queued-comment-1", steeredIntoRunId: "run-active-1", queueState: undefined, + })); const failure = new ApiError("This runner does not support steering", 409, { code: "steering_unsupported" }); await act(async () => { rejectSteer(failure); await pending; }); expect(await pending).toBe(failure); - expect((mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props).queuedCommentQueue?.entries).toHaveLength(1); + await waitForAssertion(() => { + expect((mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props).queuedCommentQueue?.entries).toHaveLength(1); + }); + expect(mockPushToast).not.toHaveBeenCalled(); }); it("promotes a steered message immediately while its durable timeline position refreshes", async () => { @@ -5975,14 +5983,16 @@ describe("IssueDetail", () => { mockActivityApi.forIssue.mockReturnValue(activityRefresh); let steeringPromise!: Promise; + let deliverySettled = false; await act(async () => { - steeringPromise = steer("queued-comment-1", queue.revision); + steeringPromise = steer("queued-comment-1", queue.revision).then(() => { deliverySettled = true; }); await Promise.resolve(); }); await waitForAssertion(() => { expect(mockIssuesApi.steerQueuedComment).toHaveBeenCalled(); expect(mockActivityApi.forIssue.mock.calls.length).toBeGreaterThan(1); expect(mockIssueChatThreadRender.mock.calls.at(-1)?.[0]?.queuedCommentQueue).toBeNull(); + expect(deliverySettled).toBe(true); }); const whileRefreshing = mockIssueChatThreadRender.mock.calls.at( @@ -6045,6 +6055,82 @@ describe("IssueDetail", () => { }); }); + it.each([ + ["success", true], ["failure", true], ["success", false], ["failure", false], + ] as const)("optimistically promotes the interrupt queue with no toast on %s (history loaded: %s)", async (outcome, historyLoaded) => { + const queue = createQueuedCommentQueue({ protocol: "legacy", steeringDisposition: "unsupported" }); + queue.entries.push({ ...queue.entries[0], position: 1, comment: createIssueComment({ id: "queued-comment-2", body: "Second direction" }) }); + mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", executionRunId: "run-active-1" })); + mockAgentsApi.list.mockResolvedValue([createAgent({ adapterType: "codex_local" })]); + mockIssuesApi.listComments.mockResolvedValue(historyLoaded ? queue.entries.map(entry => entry.comment) : []); + mockIssuesApi.getQueuedComments.mockResolvedValue(queue); + mockHeartbeatsApi.activeRunForIssue.mockResolvedValue({ + id: "run-active-1", runtimeMode: "legacy", status: "running", invocationSource: "issue", + startedAt: "2026-04-21T00:00:00.000Z", finishedAt: null, createdAt: "2026-04-21T00:00:00.000Z", + agentId: "agent-1", agentName: "Coder", adapterType: "codex_local", issueId: "issue-1", + }); + const delivery = createDeferred(); + mockIssuesApi.interruptQueuedComments.mockReturnValue(delivery.promise); + await act(async () => { root.render(); }); + type Props = { onInterruptQueued: (runId: string | null) => Promise; queuedCommentQueue: IssueQueuedCommentQueue | null; comments: Array<{ id: string; queueState?: string; conversationAnchorAt?: string }> }; + let props!: Props; + await waitForAssertion(() => { + props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props; + expect(props.queuedCommentQueue?.entries).toHaveLength(2); + }); + let pending!: Promise; + await act(async () => { pending = props.onInterruptQueued("run-active-1").catch(error => error); }); + const whilePending = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props; + expect(whilePending.queuedCommentQueue).toBeNull(); + for (const entry of queue.entries) expect(whilePending.comments).toContainEqual(expect.objectContaining({ + id: entry.comment.id, queueState: undefined, conversationAnchorAt: expect.any(String), + })); + expect(mockPushToast).not.toHaveBeenCalled(); + await act(async () => { + if (outcome === "failure") delivery.reject(new Error("Connection lost")); + else delivery.resolve(createQueuedCommentQueue({ entries: [] })); + await pending; + }); + await waitForAssertion(() => { + const final = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props; + if (outcome === "failure") expect(final.queuedCommentQueue?.entries).toHaveLength(2); + else expect(final.queuedCommentQueue).toBeNull(); + }); + expect(mockPushToast).not.toHaveBeenCalled(); + }); + + it("restores a failed classic interrupt inline without a toast or an unhandled rejection", async () => { + mockInstanceSettingsApi.getExperimental.mockResolvedValue({ enableStreamlinedUi: true, enableClassicTaskInterface: true }); + mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", executionRunId: "run-active-1" })); + mockAgentsApi.list.mockResolvedValue([createAgent({ adapterType: "codex_local" })]); + mockIssuesApi.listComments.mockResolvedValue([createIssueComment({ id: "queued-comment-1", body: "Next direction", createdAt: new Date("2026-04-21T00:00:05.000Z") })]); + mockHeartbeatsApi.activeRunForIssue.mockResolvedValue({ + id: "run-active-1", runtimeMode: "legacy", status: "running", invocationSource: "issue", + startedAt: "2026-04-21T00:00:00.000Z", finishedAt: null, createdAt: "2026-04-21T00:00:00.000Z", + agentId: "agent-1", agentName: "Coder", adapterType: "codex_local", issueId: "issue-1", + }); + mockIssuesApi.getQueuedComments.mockResolvedValue(createQueuedCommentQueue({ protocol: "legacy", steeringDisposition: "unsupported" })); + const delivery = createDeferred(); + mockIssuesApi.interruptQueuedComments.mockReturnValue(delivery.promise); + await act(async () => { root.render(); }); + type Props = { onInterruptQueued: (runId: string | null) => Promise; comments: Array<{ id: string; queueState?: string }> }; + let props!: Props; + await waitForAssertion(() => { + props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as Props; + expect(props.comments).toContainEqual(expect.objectContaining({ id: "queued-comment-1", queueState: "queued" })); + }); + let pending!: Promise; + await act(async () => { pending = props.onInterruptQueued("run-active-1").catch(error => error); }); + expect(mockIssueChatThreadRender.mock.calls.at(-1)?.[0]?.comments).toContainEqual(expect.objectContaining({ id: "queued-comment-1", queueState: undefined })); + await act(async () => { delivery.reject(new Error("Connection lost")); await pending; }); + expect(await pending).toBeUndefined(); + await waitForAssertion(() => { + expect(container.textContent).toContain("Couldn’t interrupt. Message is still queued."); + expect(mockIssueChatThreadRender.mock.calls.at(-1)?.[0]?.comments).toContainEqual(expect.objectContaining({ id: "queued-comment-1", queueState: "queued" })); + }); + expect(mockPushToast).not.toHaveBeenCalled(); + }); + it("keeps an unacknowledged queue visible while withholding server controls", async () => { mockIssuesApi.get.mockResolvedValue( createIssue({ diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index dff1a70b223..87fb4dc718f 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -1536,15 +1536,30 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ const [localSteeringPlacements, setLocalSteeringPlacements] = useState< ReadonlyMap< string, - { targetRunId: string; anchorAt: string; sequence: number } + { + targetRunId: string; + anchorAt: string; + sequence: number; + comment?: IssueDetailComment; + } > >(() => new Map()); + const [localInterruptPlacements, setLocalInterruptPlacements] = useState< + ReadonlyMap< + string, + { anchorAt: string; sequence: number; comment?: IssueDetailComment } + > + >(() => new Map()); + const [classicQueuedDeliveryError, setClassicQueuedDeliveryError] = + useState(null); useEffect(() => { setConsumedQueuedCommentIds(new Set()); setDiscardedQueuedCommentIds(new Set()); }, [issueId]); useEffect(() => { setLocalSteeringPlacements(new Map()); + setLocalInterruptPlacements(new Map()); + setClassicQueuedDeliveryError(null); }, [issueId]); const hasLiveRuns = liveRunCount > 0 || !!resolvedActiveRun; const { @@ -1757,7 +1772,20 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ } } - const projectedComments = comments.map((comment) => { + // The queue can contain a saved message outside the loaded history page. + // Keep its optimistic bubble until the canonical comment is available. + const commentsToProject = [...comments]; + const loadedCommentIds = new Set(comments.map((comment) => comment.id)); + for (const placement of [ + ...localSteeringPlacements.values(), + ...localInterruptPlacements.values(), + ]) { + if (placement.comment && !loadedCommentIds.has(placement.comment.id)) { + commentsToProject.push(placement.comment); + loadedCommentIds.add(placement.comment.id); + } + } + const projectedComments = commentsToProject.map((comment) => { const activityMeta = runMetaByCommentId.get(comment.id); // Internal run finalization can persist a reply without a separate // comment_added activity row. Its durable authoring run is stronger @@ -1985,6 +2013,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ interactions, liveRunIds, localSteeringPlacements, + localInterruptPlacements, locallyQueuedCommentRunIds, queuedCommentReason, resolvedActivity, @@ -2021,8 +2050,8 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ return targetRunId ? [{ comment, targetRunId }] : []; }); const fallbackProtocol = - liveRuntimeRun?.runtimeMode === "native" && - liveRuntimeRun.adapterType === "paperclip_runner" + liveRuntimeRun?.adapterType === "paperclip_runner" || + (!liveRuntimeRun && assigneeUsesPaperclipRunner) ? "paperclip_runner_v1" : "legacy"; return mergePendingIssueQueuedComments({ @@ -2038,6 +2067,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ discardedQueuedCommentIds, issueId, liveRuntimeRun, + assigneeUsesPaperclipRunner, locallyQueuedCommentRunIds, queuedCommentQueueEnabled, ]); @@ -2049,6 +2079,12 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ return [ { ...comment, + ...(!comment.consumedByRunId && localInterruptPlacements.has(comment.id) + ? { + conversationAnchorAt: localInterruptPlacements.get(comment.id)!.anchorAt, + conversationAnchorSequence: localInterruptPlacements.get(comment.id)!.sequence, + } + : {}), clientStatus: undefined, queueState: undefined, queueTargetRunId: null, @@ -2056,7 +2092,12 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ }, ]; }), - [commentsWithRunMeta, consumedQueuedCommentIds, discardedQueuedCommentIds], + [ + commentsWithRunMeta, + consumedQueuedCommentIds, + discardedQueuedCommentIds, + localInterruptPlacements, + ], ); const storeQueuedCommentQueue = useCallback( @@ -2146,6 +2187,23 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ throw new Error( "The queued message no longer has an active run target.", ); + const anchorAt = new Date().toISOString(); + setLocalSteeringPlacements((current) => { + const next = new Map(current); + const sequence = [...current.values()].filter( + (placement) => placement.targetRunId === targetRunId, + ).length; + next.set(commentId, { + targetRunId, + anchorAt, + sequence, + comment: effectiveQueuedCommentQueue?.entries.find( + (entry) => entry.comment.id === commentId, + )?.comment, + }); + return next; + }); + setConsumedQueuedCommentIds((current) => new Set(current).add(commentId)); try { const nextQueue = await issuesApi.steerQueuedComment( issueId, @@ -2156,23 +2214,10 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ revision, }, ); - // Keep the queue component mounted until the server accepts steering: - // its pending/error state must survive a rejected last-row action. - const anchorAt = new Date().toISOString(); - setLocalSteeringPlacements((current) => { - const next = new Map(current); - const sequence = [...current.values()].filter( - (placement) => placement.targetRunId === targetRunId, - ).length; - next.set(commentId, { targetRunId, anchorAt, sequence }); - return next; - }); - setConsumedQueuedCommentIds((current) => new Set(current).add(commentId)); - // The local steering placement already promoted the message into the - // active turn. Refresh its durable acknowledgement before publishing - // the returned queue so the local and server anchors hand off without a - // bubble-to-queue-to-bubble jump. - await Promise.all([ + // The local placement bridges the durable receipt refresh. Release the + // control as soon as delivery is acknowledged, even on a slow refetch. + storeQueuedCommentQueue(nextQueue); + void Promise.all([ queryClient.invalidateQueries({ queryKey: queryKeys.issues.comments(issueId), }), @@ -2180,7 +2225,6 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ queryKey: queryKeys.issues.activity(issueQueryRef), }), ]); - storeQueuedCommentQueue(nextQueue); } catch (error) { setConsumedQueuedCommentIds((current) => { const next = new Set(current); @@ -2196,8 +2240,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ } }, [ - effectiveQueuedCommentQueue?.queueId, - effectiveQueuedCommentQueue?.targetRunId, + effectiveQueuedCommentQueue, issueId, issueQueryRef, queryClient, @@ -2206,6 +2249,53 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ ], ); + const interruptQueuedComments = useCallback( + async (runId: string | null) => { + const queuedComments = effectiveQueuedCommentQueue?.targetRunId === runId + ? effectiveQueuedCommentQueue.entries.map((entry) => entry.comment) + : commentsWithRunMeta.filter( + (comment) => comment.queueState === "queued" && comment.queueTargetRunId === runId, + ); + const ids = queuedComments.map((comment) => comment.id); + const anchorAt = new Date().toISOString(); + setClassicQueuedDeliveryError(null); + setConsumedQueuedCommentIds((current) => new Set([...current, ...ids])); + setLocalInterruptPlacements((current) => { + const next = new Map(current); + queuedComments.forEach((comment, sequence) => { + next.set(comment.id, { anchorAt, sequence, comment }); + }); + return next; + }); + try { + await onInterruptQueued(runId); + } catch (error) { + setConsumedQueuedCommentIds((current) => { + const next = new Set(current); + for (const id of ids) next.delete(id); + return next; + }); + setLocalInterruptPlacements((current) => { + const next = new Map(current); + for (const id of ids) next.delete(id); + return next; + }); + if (classicTaskInterfaceEnabled) { + setClassicQueuedDeliveryError("Couldn’t interrupt. Message is still queued."); + return; + } + await refreshQueueAfterConflict(error); + } + }, + [ + effectiveQueuedCommentQueue, + commentsWithRunMeta, + onInterruptQueued, + refreshQueueAfterConflict, + classicTaskInterfaceEnabled, + ], + ); + const discardQueuedComment = useCallback( async (commentId: string, revision: string) => { const queueId = effectiveQueuedCommentQueue?.queueId; @@ -2445,7 +2535,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ onReviewConversation={onReviewConversation} imageUploadHandler={onImageUpload} onAttachImage={onAttachImage} - onInterruptQueued={onInterruptQueued} + onInterruptQueued={interruptQueuedComments} queuedCommentQueue={effectiveQueuedCommentQueue} onEditQueuedComment={editQueuedComment} onReorderQueuedComments={reorderQueuedComments} @@ -2495,7 +2585,14 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ tryAgainNoLiveExecutionPathPending={ tryAgainNoLiveExecutionPathPending } - footer={footer} + footer={classicQueuedDeliveryError ? ( + <> +

+ {classicQueuedDeliveryError} +

+ {footer} + + ) : footer} externalReferences={externalReferences} linkCaseReferences={linkCaseReferences} /> @@ -5110,23 +5207,10 @@ export function TaskDetailSurface({ conversation, tasksTab }: { tasksTab?: TaskS onSuccess: () => { invalidateIssueDetail(); invalidateIssueRunState(); - pushToast({ - title: "Interrupt requested", - body: "Queued messages will be sent when the previous run has stopped.", - tone: "success", - }); }, - onError: (err) => { + onError: () => { invalidateIssueDetail(); invalidateIssueRunState(); - pushToast({ - title: "Interrupt failed", - body: - err instanceof Error - ? err.message - : "Unable to interrupt the active run", - tone: "error", - }); }, });