diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 2c7535a7fcb..b25ba1962f6 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -632,7 +632,7 @@ A board comment can be an interrupt, an ownership change, both, or neither. Pape An interrupt stops the current live execution path for the issue. It does not, by itself, select the next owner. If an active run is interrupted by the board, the run may still terminate with the underlying `cancelled` status, but the issue activity and wake context should make the operator intent visible as an interruption rather than an unexplained runtime failure. -For legacy runners, **Interrupt** on a queued message stops the active run and explicitly continues the pending queue after execution cleanup. It validates the queue revision and target run, then dispatches the requested queue’s current message bodies in their saved order. Other actors’ queues cannot consume that interrupt. The persisted interrupt intent is retried by the scheduler after a promotion error or server restart until that queue is dispatched or discarded. Edits and discards remain authoritative until dispatch; deleting the final message must not create an empty continuation. Pending messages remain visible after a run stops. Cancelling only the run preserves the queue for a later explicit wake; pausing the task retains its separate queue-cancellation behavior. Native same-turn steering keeps its separate acknowledgement protocol. After an acknowledged native Stop, a follow-up saved before the stopped run finished remains pending. It cannot authorize an automatic new turn. A later user message can authorize one continuation that retains the earlier undelivered input. Closing the stopped execution coordinator also wakes the existing accounting worker in the same transaction. Complete usage can settle once; missing or partial receipts retain their accounting hold. Legacy Codex uses Ctrl-C to stop its tool sessions and cannot retry a missing-session fallback after the provider has confirmed that the session started. +For legacy runners, **Interrupt** on a queued message stops the active run and explicitly continues the pending queue after execution cleanup. It validates the queue revision and target run, then dispatches the requested queue’s current message bodies in their saved order. Other actors’ queues cannot consume that interrupt. The persisted interrupt intent is retried by the scheduler after a promotion error or server restart until that queue is dispatched or discarded. Edits and discards remain authoritative until dispatch; deleting the final message must not create an empty continuation. Pending messages remain visible after a run stops. Cancelling only the run preserves the queue for a later explicit wake; pausing the task retains its separate queue-cancellation behavior. Native same-turn steering keeps its separate acknowledgement protocol. After an acknowledged native Stop, a follow-up saved before the stopped run finished remains pending. It cannot authorize an automatic new turn. A later user message can authorize one continuation that retains the earlier undelivered input. A stopped queue also offers **Send**, which records fresh board intent for the existing messages without copying them or changing their authors. Native Send intent must be later than the stopped execution; it preserves all provider-stop, workspace, approval, pause and budget gates. It does not interrupt a live native turn. Closing the stopped execution coordinator also wakes the existing accounting worker in the same transaction. Complete usage can settle once; missing or partial receipts retain their accounting hold. Legacy Codex uses Ctrl-C to stop its tool sessions and cannot retry a missing-session fallback after the provider has confirmed that the session started. An ownership change selects who owns the issue after the comment is committed: @@ -1824,6 +1824,14 @@ comments already delivered to the failed run. A plain operator Stop still requires a new user action. The successor guard is scoped to the same agent so another agent's review participation keeps its independent recovery path. +Agent Chat `/new` can clear provider context without launching a provider. It +cannot certify that an older native provider stopped. Native execution recovery +records remain eligible across that boundary, so the next message must still +pass the existing process, controller, workspace and lease cleanup checks. A +verified stop can retire the hold for new input; prior outcomes stay recorded. +Send also checks the native predecessor when no recovery record exists. A later +context-reset record cannot hide that predecessor or its stop requirements. + An explicit queued-message Interrupt also grants one scoped cleanup retry for the stopped run. Old ephemeral leases whose cleanup predates provider stop receipts are rechecked through the recorded provider teardown path. Retained @@ -2163,13 +2171,33 @@ message posted after the run finished can authorize a fresh turn through the same continuation checks as a new message. A message from before the Stop cannot authorize that continuation by itself. Previously delivered messages stay excluded. +The Send click also covers saved question answers and confirmation responses. +Admission checks the last native Stop's process, workspace and cleanup receipts +even when no recovery action exists. Its continuation authority is recorded on +the exact queue receipt and verified again at dispatch. A deferred response keeps +its messages and controls visible and explains the current wait. Historical +recovery actions cannot stand in for the newest failed or cancelled native run: +Send checks that run's controller, process, lease and cleanup proofs as well. + +A live plan Interrupt binds the click to its target run and the exact operator +cancellation receipt. Its click may precede that run's stop time, but cannot +bypass process or workspace cleanup. Other Send clicks must follow the Stop. +Only a persisted Agent Chat `/new` authored by the requesting user retains the +conversation-reset exception. Ordinary follow-ups still wait for an accepted +result to finish committing, even after workspace export recovery clears its +execution blocker. Committed successful results no longer create continuation +waits from recovered-export history. Applied conversation resets retire older +non-workspace holds; active workspace holds remain authoritative across the +boundary. Workspace recovery holds still block the reset. Queue authorization +is read again at dispatch for both cancelled and failed native predecessors. + 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 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 +Rejected or still-deferred 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 diff --git a/maps/agent-access-map/steering.md b/maps/agent-access-map/steering.md index a58d0253850..0655c5b7955 100644 --- a/maps/agent-access-map/steering.md +++ b/maps/agent-access-map/steering.md @@ -119,7 +119,7 @@ API: `present` — representative operation IDs: Agent access: `restricted` — Board-user interruption/cancel API. Policy: Queue inspection follows task visibility; queued user-message mutation and run -cancellation require a board actor/user. +cancellation require a board actor/user. The same queue endpoint accepts an explicit **Send** for a stopped native queue. It records fresh board intent, retains original comment authors, and keeps execution cleanup and admission gates. Agent Chat `/new` also retains native process, controller, workspace and lease checks before subsequent messages execute. It does not grant agents the ability to send saved user messages or interrupt live native turns. Evidence: `source-only` — [implementation](../../server/src/routes/issues.ts#L16166) at the reviewed commit above. diff --git a/maps/feature-map/steering.md b/maps/feature-map/steering.md index cf2c7de0ecc..3adb8f1f748 100644 --- a/maps/feature-map/steering.md +++ b/maps/feature-map/steering.md @@ -35,7 +35,7 @@ Agent access: [companion](../agent-access-map/steering.md). Open an assigned task (`/issues/:issueId`). Start work, then send follow-ups while the agent runs. Use the queued-message controls to edit, reorder, discard, -steer, or interrupt when offered. Use the composer stop/pause control separately. +steer, or interrupt when offered. After Stop, choose **Send** on a saved queue to explicitly continue it without recreating the messages or attachments. Native steering still requires an active turn; Send after Stop is a separate user decision. Use the composer stop/pause control separately. Both task presentations need attention when changing their shared behavior. ### `agent-chat` diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index e02d9708a38..1d0467105af 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -739,15 +739,18 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running"); }); - it.each([null, "stopped-target", "system-receipt"])("sends a stopped legacy queue once with target %s", async (target) => { + it.each([ + { target: null, runtime: "legacy" }, { target: "stopped-target", runtime: "legacy" }, + { target: "system-receipt", runtime: "legacy" }, { target: null, runtime: "native" }, + ] as const)("sends a stopped $runtime queue once with target $target", async ({ target, runtime }) => { const seeded = await seedQueue(); if (target === "system-receipt") await db.update(agentWakeupRequests).set({ requestedByActorType: "system", requestedByActorId: "heartbeat", }).where(eq(agentWakeupRequests.id, seeded.wakeId)); - await db.update(agents).set({ adapterType: "claude_local", + await db.update(agents).set({ adapterType: runtime === "native" ? "paperclip_runner" : "claude_local", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } }, }).where(eq(agents.id, seeded.agentId)); - await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "succeeded", + await db.update(heartbeatRuns).set({ runtimeMode: runtime, status: "succeeded", finishedAt: new Date("2026-08-22T15:03:00.000Z"), }).where(eq(heartbeatRuns.id, seeded.runId)); await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId)); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 9c4beef7251..7cf93d0e9f2 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -16534,7 +16534,9 @@ export function issueRoutes( }); assertQueueMutationTarget({ queue: locked.queue, queueId: req.body.queueId, revision: req.body.revision }); const freshResponse = locked.queue.entries.some(entry => entry.source?.requiresFreshSession); - if ((locked.queue.protocol !== "legacy" && !freshResponse) || locked.state !== "deferred" || + const stoppedNativeQueue = locked.queue.protocol === "paperclip_runner_v1" && + (!locked.activeRun || ["succeeded", "failed", "timed_out", "interrupted", "cancelled"].includes(locked.activeRun.status)); + if ((locked.queue.protocol !== "legacy" && !freshResponse && !stoppedNativeQueue) || locked.state !== "deferred" || !locked.queue.entries.length || (locked.activeRun && locked.activeRun.agentId !== locked.wake.agentId)) { throw conflict("This queue does not support interruption"); @@ -16551,6 +16553,8 @@ 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" + ? { targetRunId: locked.activeRun.id } : {}), } }, updatedAt: new Date(), }).where(eq(agentWakeupRequests.id, locked.wake.id)); diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 3ee94aedb72..61d6ded503b 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -105,6 +105,13 @@ export function executionBlockerPredicate() { ); } +export function nativeExecutionRecoveryActionPredicate() { + return sql`exists (select 1 from ${heartbeatRuns} + where ${heartbeatRuns.companyId} = ${issueRecoveryActions.companyId} + and ${heartbeatRuns.id}::text = coalesce(${issueRecoveryActions.evidence}->>'runId', ${issueRecoveryActions.evidence}->>'sourceRunId') + and ${heartbeatRuns.runtimeMode} = 'native')`; +} + export async function getExecutionBlocker(db: Db, companyId: string, issueId: string, options?: { conversationResetCommentId?: string | null }): Promise { const [conversation] = await db.select({ agentId: issues.conversationAgentId, boundaryId: issues.conversationBoundaryCommentId, status: issues.status, @@ -145,7 +152,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), executionBlockerPredicate(), - boundary ? gt(issueRecoveryActions.createdAt, boundary.createdAt) : undefined, + boundary ? or(gt(issueRecoveryActions.createdAt, boundary.createdAt), nativeExecutionRecoveryActionPredicate()) : undefined, )).orderBy(desc(issueRecoveryActions.updatedAt), desc(issueRecoveryActions.id)).limit(1); if (!action) return null; const parsedRunId = z.string().guid().safeParse(action.evidence.runId ?? action.evidence.sourceRunId); diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index 378e7f1382c..616b08b3249 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -14,6 +14,7 @@ import type { ExecutionContinuationEnvelope } from "@paperclipai/shared"; import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js"; import { hasConversationContinuationPolicy } from "./conversation-continuation.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; +import { queuedInteractionId } from "./queued-interaction-response.js"; import { childReviewOutcomes } from "./native-runtime/child-review-outcomes.js"; import { isCompletedOnboardingHandoffWake } from "./chat-completion-delivery.js"; import { githubRecoveryReceipts } from "./chat-channels/github/recovery-receipts.js"; @@ -416,7 +417,15 @@ export async function buildExecutionContinuation(input: { // Admission records the board operator's authority separately from the // message author. At dispatch, prove that exact queue was adopted by this // run; caller-supplied continuation context cannot grant this authority. - const continuationAuthorizations = reconciliations.map(row => object(row.evidence.explicitUserContinuation)) + const queueIntentId = z.string().guid().safeParse(explicitContinuation.queuedCommentInterruptId); + const nativeQueuedContinuation = predecessor?.runtimeMode === "native" && ["cancelled", "failed"].includes(predecessor.status); + const [recordedQueue] = nativeQueuedContinuation && input.runId && queueIntentId.success ? await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, queueIntentId.data), eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.status, "coalesced"), + eq(agentWakeupRequests.runId, input.runId), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + )) : []; + const continuationAuthorizations = [...reconciliations.map(row => object(row.evidence.explicitUserContinuation)), + ...(recordedQueue ? [object(object(recordedQueue.payload?.queuedCommentInterrupt).continuation)] : [])] .filter(value => value.previousRunId === explicitUserSource && (!input.runId || value.runId === input.runId) && value.commentId === explicitContinuation.commentId && @@ -451,6 +460,12 @@ export async function buildExecutionContinuation(input: { ? value.failedRunId === failedRunId && retryWakes.some(wake => wake.runId === value.runId && wake.requestedByActorId === value.actorId && priorRuns.some(run => run.id === wake.runId && run.retryOfRunId === failedRunId)) + : value.interactionId + ? interactions.some(interaction => interaction.id === value.interactionId && interaction.id === triggerInteraction?.id && + ["accepted", "answered", "rejected"].includes(interaction.status) && + interruptQueues.some(queue => queue.id === value.queuedCommentInterruptId && queue.runId === value.runId && + object(queue.payload?.queuedCommentInterrupt).actorId === value.actorId && + queuedInteractionId(queue.payload) === interaction.id && queue.payload?.interactionStatus === interaction.status)) : rows.some(comment => comment.id === value.commentId && comment.authorType === "user" && (!("commentUpdatedAt" in value || "commentBodyHash" in value || value.automaticRetry) || ( diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 4608ebac4da..ce0880a4175 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -400,6 +400,200 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes for (const identity of identities) await db.delete(environmentLeases).where(eq(environmentLeases.id, identity.id)); } }); + it.each(["fresh", "predates_stop", "malformed", "wrong_actor", "provider_alive", "identity_missing", "no_click"])("requires fresh explicit Send intent for a stopped native queue: %s", async (kind) => { + const f = await seed(), queueId = randomUUID(); + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "cancelled", + processPid: kind === "provider_alive" ? process.pid : kind === "identity_missing" ? null : 999999999, + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueComments).set({ authorUserId: "original-author", createdAt: new Date("2026-09-11T09:00:00Z") }) + .where(eq(issueComments.id, f.commentId)); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution", + requestedByActorType: "system", requestedByActorId: "heartbeat", + payload: { issueId: f.issueId, _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, + ...(kind === "no_click" ? {} : { queuedCommentInterrupt: { + actorId: kind === "wrong_actor" ? "someone-else" : "board", + requestedAt: kind === "malformed" ? "invalid" : kind === "predates_stop" ? "2026-09-11T09:30:00Z" : "2026-09-11T11:00:00Z", + } }), + }, + }); + const blocked: string[] = []; + const result = await db.transaction(async tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, + queuedCommentInterruptId: queueId, dryRun: true, onBlocked: reason => blocked.push(reason) })); + if (kind === "fresh") expect(result, JSON.stringify(blocked)).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId }); + else expect(result).toBeNull(); + expect((await db.select().from(issueComments).where(eq(issueComments.id, f.commentId)))[0].authorUserId).toBe("original-author"); + }); + it.each(["clean", "pending", "live_process", "controller"])("checks the latest native Stop despite an older resolved recovery action: %s", async kind => { + const f = await seed(), latestId = randomUUID(), queueId = randomUUID(); + await db.update(heartbeatRuns).set({ createdAt: new Date("2026-09-11T08:00:00Z") }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.insert(heartbeatRuns).values({ id: latestId, companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.issueId, + runtimeMode: "native", status: "cancelled", errorCode: "cancelled", processPid: kind === "live_process" ? process.pid : 999999999, + contextSnapshot: { issueId: f.issueId }, createdAt: new Date("2026-09-11T10:15:00Z"), finishedAt: new Date("2026-09-11T10:30:00Z") }); + await db.insert(nativeRunFinalizations).values({ runId: latestId, companyId: f.companyId, issueId: f.issueId, + phase: "terminal_failure", leaseOwner: kind === "controller" ? "active-controller" : null }); + const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: latestId, provider: kind === "live_process" ? "local" : "daytona", providerLeaseId: randomUUID() }; + await db.insert(environmentLeases).values({ ...lease, status: kind === "pending" ? "pending_cleanup" : "released", releasedAt: new Date(), + cleanupStatus: kind === "pending" ? "failed" : "success", metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) } }); + 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, _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, + queuedCommentInterrupt: { actorId: "board", requestedAt: "2026-09-11T11:30:00Z" } } }); + const result = await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, queuedCommentInterruptId: queueId, dryRun: true })); + if (kind === "clean") expect(result).toMatchObject({ previousRunId: latestId }); + else expect(result).toBeNull(); + await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id)); + }); + + it.each(["clean", "pending", "missing_receipt", "wrong_resource", "controller", "workspace_owner", "cleanup_in_flight"])("checks native Stop cleanup without a recovery action: %s", async kind => { + const f = await seed(), queueId = randomUUID(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "cancelled" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "workspace_owner") await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: "other-controller" } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (kind === "controller") await db.update(nativeRunFinalizations).set({ leaseOwner: "other-controller" }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId, + provider: "daytona", providerLeaseId: randomUUID() }; + await db.insert(environmentLeases).values({ ...lease, status: kind === "pending" ? "pending_cleanup" : "released", + releasedAt: new Date(), cleanupStatus: kind === "pending" ? "failed" : "success", metadata: kind === "missing_receipt" ? {} : { + ...(kind === "cleanup_in_flight" ? { pendingCleanupInFlight: true } : {}), + remoteExecutionTermination: remoteTerminationReceipt(lease, { + providerLeaseId: kind === "wrong_resource" ? randomUUID() : lease.providerLeaseId, state: "destroyed", + }), + } }); + 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, _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, + queuedCommentInterrupt: { actorId: "board", requestedAt: "2026-09-11T11:30:00Z" } }, + }); + const blocked: string[] = []; + const result = await db.transaction(async tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, + queuedCommentInterruptId: queueId, dryRun: true, onBlocked: reason => blocked.push(reason) })); + if (kind === "clean") expect(result).toMatchObject({ previousRunId: f.sourceRunId }); + else { expect(result).toBeNull(); expect(blocked).not.toEqual([]); } + await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id)); + }); + it.each([true, false])("keeps a native Send behind cleanup through public admission without a recovery action: %s", async pending => { + const f = await seed(), queueId = randomUUID(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "cancelled" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId, provider: "daytona", providerLeaseId: randomUUID() }; + await db.insert(environmentLeases).values({ ...lease, status: pending ? "pending_cleanup" : "released", releasedAt: new Date(), + cleanupStatus: pending ? "failed" : "success", metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) } }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running", contextSnapshot: { issueId: randomUUID() } }); + await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId, + source: "on_demand", reason: "issue_commented", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: "board", + payload: { issueId: f.issueId, _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, + queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } }, + }); + await heartbeatService(db).resumeQueuedCommentInterrupt(f.companyId, queueId); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)); + if (pending) expect(receipt).toMatchObject({ status: "deferred_issue_execution", runId: null, payload: { executionWait: { reason: "remote_cleanup" } } }); + else { + expect(receipt.status).toBe("coalesced"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)))[0].contextSnapshot) + .toMatchObject({ previousRunId: f.sourceRunId }); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)); + const envelope = await buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: run.id, context: run.contextSnapshot!, summary: null, exposeLowTrustRaw: false }); + expect(envelope.interruptedRunId).toBe(f.sourceRunId); + expect(envelope.originCommentIds).toContain(f.commentId); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), + status: "resolved", outcome: "recovered", evidence: { runId: f.sourceRunId }, nextAction: "Recovery is complete." }); + const afterRecovery = await buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: run.id, context: run.contextSnapshot!, summary: null, exposeLowTrustRaw: false }); + expect(afterRecovery.originCommentIds).toContain(f.commentId); + await expect(buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: randomUUID(), context: run.contextSnapshot!, summary: null, exposeLowTrustRaw: false })) + .rejects.toThrow("continuation_user_authorization_missing"); + await db.update(agentWakeupRequests).set({ payload: { ...receipt.payload, + queuedCommentInterrupt: { ...(receipt.payload!.queuedCommentInterrupt as Record), actorId: "someone-else" }, + } }).where(eq(agentWakeupRequests.id, queueId)); + await expect(buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: run.id, context: run.contextSnapshot!, summary: null, exposeLowTrustRaw: false })) + .rejects.toThrow("continuation_user_authorization_missing"); + } + await db.delete(environmentLeases).where(eq(environmentLeases.id, lease.id)); + }); + it.each(["ask_user_questions", "request_confirmation"] as const)("sends a saved native %s response after a fresh click", async kind => { + const f = await seed(), queueId = randomUUID(); + const [interaction] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + createdByAgentId: f.agentId, sourceRunId: f.sourceRunId, kind, status: kind === "ask_user_questions" ? "answered" : "accepted", + resolvedByUserId: "board", resolvedAt: new Date("2026-09-11T09:00:00Z"), + payload: kind === "ask_user_questions" ? { version: 1, questions: [{ id: "q1", prompt: "Continue?", selectionMode: "single", options: [{ id: "yes", label: "Yes" }] }] } + : { version: 1, prompt: "Continue?" }, + result: kind === "ask_user_questions" ? { version: 1, answers: [{ questionId: "q1", optionIds: ["yes"] }] } + : { version: 1, outcome: "accepted" }, + }).returning(); + 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: interaction.status, + queuedCommentInterrupt: { actorId: "board", requestedAt: "2026-09-11T11:30:00Z" } }, + }); + const result = await db.transaction(async tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, + commentId: null, queuedCommentInterruptId: queueId })); + expect(result).toMatchObject({ previousRunId: f.sourceRunId }); + const context = { issueId: f.issueId, interactionId: interaction.id, mutation: "interaction", interactionKind: kind, + interactionStatus: interaction.status, explicitUserContinuation: result }; + await db.insert(heartbeatRuns).values({ id: f.successorRunId, companyId: f.companyId, agentId: f.agentId, + runtimeMode: "native", nativeIssueId: f.issueId, status: "queued", contextSnapshot: context }); + await db.update(agentWakeupRequests).set({ status: "coalesced", runId: f.successorRunId }).where(eq(agentWakeupRequests.id, queueId)); + const envelope = await buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: f.successorRunId, context, summary: null, exposeLowTrustRaw: false }); + expect(envelope.interruptedRunId).toBe(f.sourceRunId); + 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 => { + const f = await seed(), queueId = randomUUID(); + const [interaction] = 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" }, + }).returning(); + await db.update(heartbeatRuns).set({ status: "cancelled", resultJson: kind === "missing_receipt" ? null : { + operatorInterrupted: true, interruptionSource: "issue_comment_interrupt", interruptedIssueId: f.issueId, + interruptedByActorType: "user", interruptedByActorId: kind === "wrong_actor" ? "someone-else" : "board", + queuedCommentInterruptQueueId: kind === "wrong_queue" ? randomUUID() : queueId, + } }).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" }, + 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 }); + else expect(result).toBeNull(); + }); + + it.each(["valid", "workspace_hold", "wrong_author"])("admits only a verified chat context reset while keeping workspace holds: %s", async scenario => { + const workspaceHold = scenario === "workspace_hold"; + const f = await seed(), settings = instanceSettingsService(db), previous = await settings.getExperimental(); + await settings.updateExperimental({ enableAgentChat: true }); + try { + await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: "board", conversationState: "active", status: "in_progress" }).where(eq(issues.id, f.issueId)); + await db.update(issueComments).set({ body: "/new" }).where(eq(issueComments.id, f.commentId)); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + if (workspaceHold) await db.update(issueRecoveryActions).set({ + evidence: { runId: f.sourceRunId, workspaceRestoreFailure: "restore_unsafe_archive", automaticRecovery: { replay: "blocked" } }, + }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + const blocker = await getExecutionBlocker(db, f.companyId, f.issueId, { conversationResetCommentId: f.commentId }); + if (workspaceHold) expect(blocker).not.toBeNull(); else expect(blocker).toBeNull(); + const wake = await heartbeatService(db).wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: scenario === "wrong_author" ? "another-user" : "board", payload: { issueId: f.issueId, commentId: f.commentId }, + contextSnapshot: { source: "issue.comment", issueId: f.issueId, wakeCommentId: f.commentId }, + }); + if (workspaceHold || scenario === "wrong_author") expect(wake).toBeNull(); else expect(wake).not.toBeNull(); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)))[0].evidence) + .not.toHaveProperty("explicitUserContinuation"); + } finally { await settings.updateExperimental(previous); } + }); + it("a durable queue interrupt authorizes older legacy messages but still requires the provider to stop", async () => { const f = await seed(); await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId)); @@ -1954,6 +2148,135 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes await settings.updateExperimental({ enableAgentChat: experimental.enableAgentChat }); } }); + it("defers an ordinary follow-up while a recovered export's successful result is still finalizing", async () => { + const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f); + await db.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, failureDetail: null }) + .where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "recovered", cause: "native_workspace_sync_out_retry_exhausted", + evidence: { runId: f.sourceRunId, workspaceRecovery: { status: "recovered" }, systemRecovery: createSystemRecovery(f.sourceRunId) }, + }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + const blocked = vi.fn(); + expect(await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, onBlocked: blocked, dryRun: true }))).toBeNull(); + expect(blocked).toHaveBeenCalledWith("finalizing_saved_result", expect.any(String)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const wake = await heartbeatService(db).wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId }, + contextSnapshot: { source: "issue.comment", issueId: f.issueId, wakeCommentId: f.commentId }, + }); + expect(wake).toBeNull(); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.nativeIssueId, f.issueId), eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(0); + const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(waiting).toMatchObject({ status: "deferred_issue_execution", runId: null, payload: { executionWait: { reason: "finalizing_saved_result" } } }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0]).toMatchObject({ phase: "result_accepted", resultId }); + }); + + it.each(["completed", "after_reset", "after_reset_workspace_hold"])("admits fresh messages after historical execution work is cleared: %s", async scenario => { + const f = await seed(), settings = instanceSettingsService(db), previous = await settings.getExperimental(); + try { + if (scenario === "completed") { + await seedResultBeforeTerminalFailure(f); + await db.update(nativeRunFinalizations).set({ phase: "committed", leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "recovered", cause: "native_workspace_sync_out_retry_exhausted", + evidence: { runId: f.sourceRunId, workspaceRecovery: { status: "recovered" }, systemRecovery: createSystemRecovery(f.sourceRunId) }, + }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + } else { + await settings.updateExperimental({ enableAgentChat: true }); + const boundaryId = randomUUID(); + await db.insert(issueComments).values({ id: boundaryId, companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "board", body: "/new", createdAt: new Date("2026-09-11T10:30:00Z") }); + await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: "board", conversationState: "active", conversationBoundaryCommentId: boundaryId }).where(eq(issues.id, f.issueId)); + await db.update(issueRecoveryActions).set({ createdAt: new Date("2026-09-11T09:00:00Z") }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); + expect(await recordNativeLocalProcessStop(db, source)).toBe(true); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (scenario === "after_reset_workspace_hold") { + await db.update(heartbeatRuns).set({ processPid: 999999999, resultJson: { workspaceRestoreFailure: "restore_unsafe_archive" } }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.update(issueRecoveryActions).set({ evidence: { runId: f.sourceRunId, workspaceRestoreFailure: "restore_unsafe_archive", automaticRecovery: { replay: "blocked" } } }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + } + } + const blocker = await getExecutionBlocker(db, f.companyId, f.issueId); + if (scenario === "completed") expect(blocker).toBeNull(); else expect(blocker).not.toBeNull(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const wake = await heartbeatService(db).wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId }, + contextSnapshot: { source: "issue.comment", issueId: f.issueId, wakeCommentId: f.commentId }, + }); + if (scenario === "after_reset_workspace_hold") expect(wake).toBeNull(); + else { expect(wake).not.toBeNull(); expect(wake).toMatchObject({ status: "queued" }); expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); } + } finally { await settings.updateExperimental(previous); } + }); + + it.each(["live_process", "missing_identity", "quarantined", "verified", "remote_cleanup", "controller"])( + "keeps native stop and cleanup checks after an applied chat reset: %s", async scenario => { + const f = await seed(), settings = instanceSettingsService(db), previous = await settings.getExperimental(); + const retire = vi.fn(() => true); + const verify = vi.spyOn(nativeExecutor, "verifyStoppedNativeSessionForContinuation").mockResolvedValue( + scenario === "verified" ? { evidence: { runId: f.sourceRunId, schema: "paperclip.stopped_native_conversation.v1" }, retire } : null); + try { + await settings.updateExperimental({ enableAgentChat: true }); + const boundaryId = randomUUID(); + await db.insert(issueComments).values({ id: boundaryId, companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "board", body: "/new", createdAt: new Date("2026-09-11T10:30:00Z") }); + await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: "board", conversationState: "active", conversationBoundaryCommentId: boundaryId }).where(eq(issues.id, f.issueId)); + await db.update(issueRecoveryActions).set({ createdAt: new Date("2026-09-11T09:00:00Z") }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(heartbeatRuns).set({ createdAt: new Date("2026-09-11T09:00:00Z"), + processPid: scenario === "live_process" ? process.pid : scenario === "missing_identity" ? null : 999999999, + errorCode: "native_session_cleanup_quarantined", + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (scenario === "remote_cleanup") await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId, provider: "daytona", providerLeaseId: "old-chat-environment", status: "active", leasePolicy: "ephemeral" }); + if (scenario === "controller") await db.update(nativeRunFinalizations).set({ leaseOwner: "old-controller" }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const wake = await heartbeatService(db).wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId }, + contextSnapshot: { source: "issue.comment", issueId: f.issueId, wakeCommentId: f.commentId }, + }); + if (scenario === "verified") { + expect(wake).toMatchObject({ status: "queued" }); + expect(retire).toHaveBeenCalledOnce(); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + } else { + expect(wake).toBeNull(); + const [saved] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), eq(agentWakeupRequests.reason, "issue_commented"))); + expect(saved).toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + expect(retire).not.toHaveBeenCalled(); + } + } finally { verify.mockRestore(); await settings.updateExperimental(previous); } + }); + + it.each(["live", "stopped"])("checks the native predecessor after reset even without a recovery record: %s", async state => { + const f = await seed(), queueId = randomUUID(), boundaryId = randomUUID(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(heartbeatRuns).set({ createdAt: new Date("2026-09-11T09:00:00Z"), processPid: state === "live" ? process.pid : 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.insert(issueComments).values({ id: boundaryId, companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: "board", body: "/new", createdAt: new Date("2026-09-11T10:30:00Z") }); + await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: "board", conversationState: "active", conversationBoundaryCommentId: boundaryId }).where(eq(issues.id, f.issueId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, runtimeMode: "legacy", status: "succeeded", createdAt: new Date("2026-09-11T10:30:01Z"), finishedAt: new Date("2026-09-11T10:30:02Z"), contextSnapshot: { issueId: f.issueId, wakeCommentId: boundaryId } }); + 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, _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, queuedCommentInterrupt: { actorId: "board", requestedAt: "2026-09-11T11:30:00Z" } } }); + const blocked: string[] = []; + const continuation = await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, queuedCommentInterruptId: queueId, onBlocked: reason => blocked.push(reason) })); + if (state === "live") { expect(continuation).toBeNull(); expect(blocked).toEqual(["process_running"]); } + else { expect(continuation).toMatchObject({ previousRunId: f.sourceRunId }); expect(blocked).toEqual([]); } + }); + + it("dispatches a fresh native Send after a committed failed result without a recovery action", async () => { + const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f), queueId = randomUUID(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(nativeRunResults).set({ resultJson: { terminal: { runTerminalState: "failed" } } }).where(eq(nativeRunResults.id, resultId)); + await db.update(nativeRunFinalizations).set({ phase: "committed", failureDetail: null, leaseOwner: null }).where(eq(nativeRunFinalizations.runId, 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, _paperclipWakeContext: { wakeCommentIds: [f.commentId] }, queuedCommentInterrupt: { actorId: "board", requestedAt: "2026-09-11T11:30:00Z" } } }); + const continuation = await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, queuedCommentInterruptId: queueId })); + expect(continuation).toMatchObject({ previousRunId: f.sourceRunId }); + const context = { issueId: f.issueId, previousRunId: f.sourceRunId, wakeCommentId: f.commentId, wakeCommentIds: [f.commentId], explicitUserContinuation: continuation }; + await db.insert(heartbeatRuns).values({ id: f.successorRunId, companyId: f.companyId, agentId: f.agentId, runtimeMode: "native", nativeIssueId: f.issueId, status: "queued", contextSnapshot: context }); + await db.update(agentWakeupRequests).set({ status: "coalesced", runId: f.successorRunId }).where(eq(agentWakeupRequests.id, queueId)); + const envelope = await buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, runId: f.successorRunId, context, summary: null, exposeLowTrustRaw: false }); + expect(envelope.interruptedRunId).toBe(f.sourceRunId); + expect(envelope.originCommentIds).toContain(f.commentId); + }); + it("keeps exhausted workspace export on its own repair path after the sandbox stops", async () => { const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f); await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index fccbe652853..fce18d379f2 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -15,13 +15,13 @@ import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recorde import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; import { z } from "zod"; -import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; +import { and, desc, eq, gt, inArray, isNull, ne, or, sql } from "drizzle-orm"; import { agents, agentWakeupRequests, approvals, issueApprovals, issueThreadInteractions, environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db, } from "@paperclipai/db"; -import { executionBlockerPredicate, getExecutionBlocker } from "./execution-blocker.js"; +import { executionBlockerPredicate, getExecutionBlocker, hasPendingRunCleanup, nativeExecutionRecoveryActionPredicate } from "./execution-blocker.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; import { adapterExecutionControls } from "./adapter-execution-control.js"; import { persistActivity } from "./activity-log.js"; @@ -124,14 +124,13 @@ export async function admitExplicitNativeContinuation(input: { reason: string | null; commentId: string | null; successorRunId: string; failedRunId?: string | null; taskRetryRequestId?: string; - /** Server-recorded board intent to send an existing legacy message queue. */ queuedCommentInterruptId?: string; /** Internal delivery of an unconsumed, user-authored saved queue entry. */ queuedCommentRequestId?: string; durableChatRequest?: DurableChatWakeupRequest; dryRun?: boolean; onBlocked?: (reason: string, message: string) => void; -}): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string; taskRetryRequestId?: string; readiness?: ContinuationReadiness } | null> { +}): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string; taskRetryRequestId?: string; queuedCommentInterruptId?: string; readiness?: ContinuationReadiness } | null> { const { db, companyId, issueId, agentId, actorId, commentId } = input; const blocked = (reason: string, message: string) => { input.onBlocked?.(reason, message); return null; }; if (input.actorType !== "user" || !actorId) return null; @@ -204,21 +203,53 @@ export async function admitExplicitNativeContinuation(input: { isNull(issueComments.deletedAt), )); if (!retry && !taskRetry && !response && !comment?.body.trim()) return null; - const authorizedAt = response?.comment.createdAt ?? comment?.createdAt ?? retryWake?.requestedAt ?? new Date(); const [agent] = await db.select().from(agents).where(and(eq(agents.companyId, companyId), eq(agents.id, agentId))); if (!agent || (!isConversationAdapter(agent.adapterType) && agent.adapterType !== "paperclip_runner")) return null; - if (queuedInterrupt && !isConversationAdapter(agent.adapterType) && !response?.source.requiresFreshSession) return null; + const agentIsNative = agent.adapterType === "paperclip_runner"; + const interruptIntent = interruptQueue?.payload?.queuedCommentInterrupt as Record | undefined; + const interruptRequestedAt = typeof interruptIntent?.requestedAt === "string" ? new Date(interruptIntent.requestedAt) : null; + const nativeQueuedSend = queuedInterrupt && agentIsNative; + if (nativeQueuedSend && (!interruptRequestedAt || !Number.isFinite(interruptRequestedAt.getTime()))) + return blocked("send_intent_missing", "Send the saved message again to continue."); + const authorizedAt = nativeQueuedSend ? interruptRequestedAt! + : response?.comment.createdAt ?? comment?.createdAt ?? retryWake?.requestedAt ?? new Date(); + const blocker = await getExecutionBlocker(db, companyId, issueId); + const [boundary] = await db.select({ createdAt: issueComments.createdAt }).from(issues).innerJoin(issueComments, and( + eq(issueComments.companyId, issues.companyId), eq(issueComments.issueId, issues.id), + eq(issueComments.id, issues.conversationBoundaryCommentId), + )).where(and(eq(issues.companyId, companyId), eq(issues.id, issueId), sql`${issues.conversationAgentId} is not null`)).limit(1); const actions = await db.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + boundary ? or(gt(issueRecoveryActions.createdAt, boundary.createdAt), nativeExecutionRecoveryActionPredicate(), + blocker?.recoveryActionId ? eq(issueRecoveryActions.id, blocker.recoveryActionId) : undefined) : undefined, or(executionBlockerPredicate(), and( inArray(issueRecoveryActions.cause, ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]), sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' = 'recovered'`, sql`${issueRecoveryActions.evidence}->'executionReconciliation' is null`, sql`${issueRecoveryActions.evidence}->'explicitUserContinuation' is null`, + sql`not exists (select 1 from ${heartbeatRuns} completed + join ${nativeRunFinalizations} finalized on finalized.run_id = completed.id and finalized.company_id = completed.company_id + join ${nativeRunResults} result on result.id = finalized.result_id and result.run_id = completed.id and result.company_id = completed.company_id + where completed.company_id = ${companyId} and result.issue_id = ${issueId} + and completed.id::text = coalesce(${issueRecoveryActions.evidence}->>'runId', ${issueRecoveryActions.evidence}->>'sourceRunId') + and completed.runtime_mode = 'native' and completed.status = 'succeeded' and finalized.phase = 'committed' + and result.schema_status = 'accepted' and result.result_json#>>'{terminal,runTerminalState}' = 'succeeded')`, )), )).for("update"); - if (!actions.length) return null; - const blocker = await getExecutionBlocker(db, companyId, issueId); + const sourceChecks = actions.map(action => ({ cause: action.cause, evidence: action.evidence })); + if (nativeQueuedSend) { + const [latest] = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), + eq(heartbeatRuns.runtimeMode, "native"), + or(eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.issueId, issueId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`))) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); + if (latest?.runtimeMode === "native" && ["cancelled", "failed"].includes(latest.status) && + !sourceChecks.some(source => (source.evidence.runId ?? source.evidence.sourceRunId) === latest.id)) { + sourceChecks.push({ cause: "native_continuation_requires_reconciliation", evidence: { runId: latest.id } }); + } + } + if (!sourceChecks.length) return null; if (blocker && blocker.recoveryActionId === null) return null; const [pendingInteraction] = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId), @@ -236,7 +267,7 @@ export async function admitExplicitNativeContinuation(input: { const freshSources = new Map(); const stoppedSessions: Array<{ evidence: Record; retire: () => boolean }> = []; const cancelledStartupIds = new Set(); - for (const action of actions) { + for (const action of sourceChecks) { const runId = action.evidence.runId ?? action.evidence.sourceRunId; if (typeof runId !== "string") return blocked("source_missing", "The stopped run could not be identified. Your message is saved."); // Text comparison keeps malformed historical evidence a hold, not a UUID cast error. @@ -292,6 +323,13 @@ export async function admitExplicitNativeContinuation(input: { if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId || lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null; run = lockedRun; + const interruptedByThisClick = response?.source.requiresFreshSession === true && + 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 && + run.resultJson?.queuedCommentInterruptQueueId === input.queuedCommentInterruptId; + if (nativeQueuedSend && authorizedAt <= run.finishedAt! && !interruptedByThisClick) + return blocked("send_predates_stop", "Send the saved message again now that the previous run has stopped."); if (coordinator && coordinator.phase !== "committed" && coordinator.resultId && readSystemRecovery(action.evidence)) { const [completedResult] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( @@ -317,6 +355,7 @@ export async function admitExplicitNativeContinuation(input: { !(queuedRequest && run.runtimeMode === "native" && (run.status !== "cancelled" || authorizedAt > run.finishedAt!)) && !(queuedRequest && cancelledStartup && authorizedAt > run.finishedAt!) && + !(nativeQueuedSend && (run.runtimeMode === "native" || cancelledStartup) && authorizedAt > run.finishedAt!) && !(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. @@ -385,6 +424,8 @@ export async function admitExplicitNativeContinuation(input: { if (!stopped) return blocked("local_cleanup", "Waiting for the previous provider and its tools to stop. Your message is saved."); stoppedSessions.push(stopped); } + if (nativeQueuedSend && await hasPendingRunCleanup(db, companyId, run.id)) + return blocked("cleanup_pending", "Waiting for the previous run to finish cleanup. Your message is saved."); if (freshIntent && !await hasVerifiedWorkspaceRepair(db, run)) { freshSources.set(run.id, freshIntent); readiness = { mode: "fresh_isolated_workspace", retryIntentId: freshIntent, sourceRunId: run.id, @@ -410,6 +451,7 @@ export async function admitExplicitNativeContinuation(input: { context: { previousRunId: previous.id, wakeCommentId: commentId }, summary: null, exposeLowTrustRaw: false }); const continuation = { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}), + ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId! } : {}), ...(taskRetry ? { taskRetryRequestId: input.taskRetryRequestId! } : {}), ...(readiness ? { readiness } : {}) }; if (input.dryRun) return continuation; for (const stopped of stoppedSessions) { @@ -427,6 +469,10 @@ export async function admitExplicitNativeContinuation(input: { ...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}), ...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId, previousRunId: previous.id, recordedAt: new Date().toISOString() }; + if (input.queuedCommentInterruptId) await db.update(agentWakeupRequests).set({ + payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{queuedCommentInterrupt,continuation}', ${JSON.stringify(authorization)}::jsonb)`, + }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.queuedCommentInterruptId), + eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"))); if (cancelledStartupIds.size) await db.transaction(async tx => { await notifyExecutionRunWork(tx); for (const runId of cancelledStartupIds) { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index b9388d1b68a..5293009a369 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -25,6 +25,7 @@ import { nonIdleSlackIssueCondition } from "../slack-conversation-state.js"; import { readQueuedInteractionResponse } from "../queued-interaction-response.js"; import { isConversation, + isConversationReset, isConversationExecutionWake, isWaitingConversation, } from "../agent-conversations.js"; @@ -3320,8 +3321,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } let continuationWait = { reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." }; + let continuationBlocked = false; const deferBlockedExecution = async ( - executionBlocker: NonNullable>>, + executionBlocker: Pick>>, "recoveryActionId" | "cause" | "nextAction">, ) => { if (caseRequest) waitForCaseAdmission("execution_reconciliation_required", executionBlocker.nextAction); opts.issueStateGuard?.onExecutionDeferred?.(executionBlocker.cause); @@ -3371,9 +3373,17 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende return { kind: "deferred" as const, admitted: true }; }; const explicitContinuationRunId = randomUUID(); + const [resetComment] = isConversation(issue) && opts.requestedByActorType === "user" && opts.requestedByActorId && wakeCommentId + ? await tx.select({ body: issueComments.body }).from(issueComments).where(and( + eq(issueComments.companyId, issue.companyId), eq(issueComments.issueId, issue.id), + eq(issueComments.id, wakeCommentId), eq(issueComments.authorUserId, opts.requestedByActorId), + eq(issueComments.authorType, "user"), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), + )).limit(1) + : []; + const verifiedConversationReset = Boolean(resetComment && isConversationReset(resetComment.body)); const executionBlocker = await getExecutionBlocker( tx as unknown as Db, issue.companyId, issue.id, - { conversationResetCommentId: opts.requestedByActorType === "user" ? wakeCommentId : null }, + { conversationResetCommentId: verifiedConversationReset ? wakeCommentId : null }, ); // Prove eligibility without retiring the hold. Later gates can still // decline this wake; hold retirement and successor creation stay atomic. @@ -3382,7 +3392,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // failed-run retry actions on their separate exact-request path. const ownsReconstructionContinuation = reconciledSourceRunId && executionBlocker?.runId === reconciledSourceRunId && executionBlocker.recoveryActionId === enrichedContextSnapshot.recoveryActionId; - if (executionBlocker && !ownsReconstructionContinuation && !(await admitExplicitNativeContinuation({ + const continuationHold = () => executionBlocker ?? { + recoveryActionId: null, cause: "native_continuation_requires_reconciliation" as const, nextAction: continuationWait.message, + }; + if ((executionBlocker || opts.queuedCommentInterruptId) && !ownsReconstructionContinuation && !(await admitExplicitNativeContinuation({ db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id, agentId, actorType: taskRetryActorId ? "user" : opts.requestedByActorType, actorId: taskRetryActorId ?? opts.requestedByActorId, reason: taskRetryActorId ? "task_retry" : durableRequest && !failedChatRetry ? "issue_commented" : reason, @@ -3392,8 +3405,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende taskRetryRequestId: opts.taskRetryRequestId, ...(durableRequest?.resumeDeferred && !failedChatRetry ? { durableChatRequest: durableRequest } : {}), dryRun: true, - onBlocked: (reason, message) => { continuationWait = { reason, message }; }, - }))) return deferBlockedExecution(executionBlocker); + onBlocked: (reason, message) => { continuationBlocked = true; continuationWait = { reason, message }; }, + })) && (executionBlocker || (continuationBlocked && !verifiedConversationReset))) return deferBlockedExecution(continuationHold()); if ( @@ -4192,8 +4205,10 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende queuedCommentRequestId: opts.queuedCommentRequestId, taskRetryRequestId: opts.taskRetryRequestId, ...(durableRequest?.resumeDeferred && !failedChatRetry ? { durableChatRequest: durableRequest } : {}), + onBlocked: (reason, message) => { continuationBlocked = true; continuationWait = { reason, message }; }, }); - if (!explicitContinuation && executionBlocker && !ownsReconstructionContinuation) return deferBlockedExecution(executionBlocker); + if (!explicitContinuation && (executionBlocker || (continuationBlocked && !verifiedConversationReset)) && !ownsReconstructionContinuation) + return deferBlockedExecution(continuationHold()); if (explicitContinuation) { enrichedContextSnapshot.forceFreshSession = true; enrichedContextSnapshot.previousRunId = explicitContinuation.previousRunId; diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 2ce3cf3c363..c1b50a46517 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -629,10 +629,10 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont } async function resumeQueuedCommentInterrupt(companyId: string, queueId: string, opts?: { retryCleanup?: boolean }) { - return resumeSavedLegacyComments(companyId, queueId, true, opts); + return resumeSavedComments(companyId, queueId, true, opts); } - async function resumeSavedLegacyComments(companyId: string, queueId: string, interrupted = false, opts?: { retryCleanup?: boolean }) { + async function resumeSavedComments(companyId: string, queueId: string, interrupted = false, opts?: { retryCleanup?: boolean }) { const [wake] = await db.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"), @@ -664,7 +664,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont } if (!actorId) return; const agent = await getAgent(wake.agentId); - if (!agent || agent.companyId !== companyId || (agent.adapterType === "paperclip_runner" && !response?.source.requiresFreshSession)) return; + if (!agent || agent.companyId !== companyId || (agent.adapterType === "paperclip_runner" && !interrupted && !response?.source.requiresFreshSession)) return; const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, wake.agentId), @@ -822,7 +822,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont const run = await getRunExecutionWait(wake.companyId, sourceId); if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue; if (canContinueCancelledRun(run)) { - await resumeSavedLegacyComments(wake.companyId, wake.id).catch(err => { + await resumeSavedComments(wake.companyId, wake.id).catch(err => { logger.warn({ err, runId: run.id }, "failed to resume saved input after provider cancellation"); }); continue; @@ -1700,7 +1700,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont cancelRunInternal, resumeExecutionWaitComments, resumeQueuedCommentInterrupt, - resumeSavedLegacyComments, + resumeSavedLegacyComments: resumeSavedComments, resumeTaskRetry, cancelActiveForAgentInternal, cancelInvocationsForAgentsInternal, diff --git a/tests/e2e/acp-stop-continuation.spec.ts b/tests/e2e/acp-stop-continuation.spec.ts index 3a5735e76b0..de927420a86 100644 --- a/tests/e2e/acp-stop-continuation.spec.ts +++ b/tests/e2e/acp-stop-continuation.spec.ts @@ -9,8 +9,13 @@ async function json(response: APIResponse) { return JSON.parse(body); } -for (const { unfinishedWrite, stopResponse } of [{ unfinishedWrite: false, stopResponse: false }, { unfinishedWrite: true, stopResponse: false }, { unfinishedWrite: false, stopResponse: true }]) { - test(`embedded ACP Stop: ${unfinishedWrite ? "Interrupt continues without replaying the write" : stopResponse ? "composer Stop preserves queued input and accepts a new direction" : "Interrupt delivers queued input in the same session"}`, async ({ page, request }) => { +for (const { unfinishedWrite, stopResponse, sendSavedQueue = false } of [ + { unfinishedWrite: false, stopResponse: false }, + { unfinishedWrite: true, stopResponse: false }, + { unfinishedWrite: false, stopResponse: true }, + { unfinishedWrite: false, stopResponse: true, sendSavedQueue: true }, +]) { + test(`embedded ACP Stop: ${unfinishedWrite ? "Interrupt continues without replaying the write" : sendSavedQueue ? "Send resumes saved input after Stop" : stopResponse ? "composer Stop preserves queued input and accepts a new direction" : "Interrupt delivers queued input in the same session"}`, async ({ page, request }) => { test.setTimeout(120_000); const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-stop-browser-")); const company = await json(await request.post("/api/companies", { data: { name: `ACP Stop ${Date.now()}` } })); @@ -64,8 +69,12 @@ for (const { unfinishedWrite, stopResponse } of [{ unfinishedWrite: false, stopR const saved = await json(await request.get(`/api/issues/${issue.id}/queued-comments`)); expect(JSON.stringify(saved.entries)).toContain("List my recent Drive files."); expect((await readFile(path.join(root, "prompts"), "utf8")).trim().split("\n")).toHaveLength(1); - await editor.fill("Please continue with the saved request."); - await page.getByRole("button", { name: "Send", exact: true }).click(); + if (sendSavedQueue) { + await page.getByRole("button", { name: "Send queued messages", exact: true }).click(); + } else { + await editor.fill("Please continue with the saved request."); + await page.getByRole("button", { name: "Send", exact: true }).click(); + } } await expect(page.getByText("Answered the pending follow-up once.", { exact: false })).toBeVisible({ timeout: 30_000 }); await expect.poll(async () => (await json(await request.get(`/api/issues/${issue.id}/live-runs`))).length).toBe(0); diff --git a/ui/src/components/IssueChatThread.tsx b/ui/src/components/IssueChatThread.tsx index 3b0b64cdfd7..a1099bc37c9 100644 --- a/ui/src/components/IssueChatThread.tsx +++ b/ui/src/components/IssueChatThread.tsx @@ -283,7 +283,7 @@ interface IssueChatMessageContext { stoppingRunLabel?: string; stopRunVariant?: "stop" | "pause"; runFinalizationActions?: readonly IssueChatRunFinalizationAction[]; - onInterruptQueued?: (runId: string | null) => Promise; + onInterruptQueued?: (runId: string | null) => Promise; onCancelQueued?: (commentId: string) => void; onDeleteComment?: (commentId: string) => Promise | void; onImageClick?: (src: string) => void; @@ -673,7 +673,7 @@ interface IssueChatThreadProps { transcriptsByRunId?: ReadonlyMap; hasOutputForRun?: (runId: string) => boolean; includeSucceededRunsWithoutOutput?: boolean; - onInterruptQueued?: (runId: string | null) => Promise; + onInterruptQueued?: (runId: string | null) => Promise; onCancelQueued?: (commentId: string) => void; /** Authoritative PRP queue. The classic thread intentionally ignores it. */ queuedCommentQueue?: IssueQueuedCommentQueue | null; diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index c58a4a635cc..b4da55dad3c 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -4337,7 +4337,7 @@ describe("TaskChatThread Paperclip Runner queue", () => { expect(occurrenceCount(queuedComment.body)).toBe(1); }); - it.each(["run-1", null])("offers legacy interruption only with a live target (%s)", (targetRunId) => { + it.each(["run-1", null])("offers Interrupt for a live legacy queue and Send for a stopped queue (%s)", (targetRunId) => { const onInterruptQueued = vi.fn(async () => {}); renderThread({ comments: [ @@ -4372,14 +4372,11 @@ describe("TaskChatThread Paperclip Runner queue", () => { const interrupt = container.querySelector( '[data-testid="task-chat-queued-interrupt-queued-prp-1"]', ); - if (targetRunId) { - expect(interrupt).not.toBeNull(); - flushSync(() => interrupt!.click()); - expect(onInterruptQueued).toHaveBeenCalledWith(targetRunId); - } else { - expect(interrupt).toBeNull(); - expect(onInterruptQueued).not.toHaveBeenCalled(); - } + expect(interrupt).not.toBeNull(); + expect(interrupt!.textContent).toBe(targetRunId ? "Interrupt" : "Send"); + expect(onInterruptQueued).not.toHaveBeenCalled(); + flushSync(() => interrupt!.click()); + expect(onInterruptQueued).toHaveBeenCalledWith(targetRunId); }); it("cancels an optimistic queued row locally before server acknowledgement", async () => { diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index ec184eb080e..7101fbdea7c 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -3281,7 +3281,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { onInterrupt={ onInterruptQueued && queuedMessageQueue?.queueId ? async () => { - await onInterruptQueued( + return onInterruptQueued( queuedMessageQueue.targetRunId, ); } diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx index 97d9b6ac003..f5284d14b8b 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx @@ -394,10 +394,35 @@ describe("TaskChatQueuedMessages", () => { ), ).toBeNull(); }); - it("hides Steer and Interrupt without a live target while leaving saved messages editable", () => { - render({ queue: { ...queue, targetRunId: null, protocol: "legacy", steeringDisposition: "unsupported" } }); + it("keeps deferred sends visible and explains their current wait", async () => { + const saved = { ...queue, targetRunId: null, executionWait: { reason: "decision_pending", message: "Resolve the pending approval before sending." } }; + const onInterrupt = vi.fn().mockResolvedValue(saved); + render({ queue: saved, onInterrupt }); + await act(async () => { container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')!.click(); }); expect(container.textContent).toContain("First queued message"); - expect(container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')).toBeNull(); + expect(container.textContent).toContain(saved.executionWait.message); + expect(container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')?.disabled).toBe(false); + }); + it("clears a deferred wait message when a refreshed queue finishes sending", async () => { + const saved = { ...queue, targetRunId: null, executionWait: { reason: "decision_pending", message: "Resolve the pending approval before sending." } }; + render({ queue: saved, onInterrupt: vi.fn().mockResolvedValue(saved) }); + await act(async () => { container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')!.click(); }); + expect(container.textContent).toContain(saved.executionWait.message); + render({ queue: { ...saved, queueId: null, entries: [], state: null, executionWait: null } }); + await act(async () => {}); + expect(container.textContent).not.toContain(saved.executionWait.message); + expect(container.querySelector('[data-testid="task-chat-queued-messages"]')).toBeNull(); + }); + it.each(["legacy", "paperclip_runner_v1"] as const)("sends a saved %s queue explicitly after Stop", async (protocol) => { + const onInterrupt = vi.fn().mockResolvedValue(undefined); + render({ queue: { ...queue, targetRunId: null, protocol, steeringDisposition: "temporarily_unavailable" }, onInterrupt }); + expect(onInterrupt).not.toHaveBeenCalled(); + const send = container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]'); + expect(send?.textContent).toBe("Send"); + expect(send?.getAttribute("aria-label")).toBe("Send queued messages"); + expect(send?.disabled).toBe(false); + await act(async () => { send!.click(); }); + expect(onInterrupt).toHaveBeenCalledOnce(); expect(container.querySelector('[data-testid="task-chat-queued-steer-comment-1"]')).toBeNull(); }); diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx index 691b4622200..4404a193e29 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx @@ -71,7 +71,7 @@ export interface TaskChatQueuedMessagesProps { onEdit: (commentId: string) => void; onReorder: (orderedCommentIds: string[], revision: string) => Promise; onSteer: (commentId: string, revision: string) => Promise; - onInterrupt?: () => Promise; + onInterrupt?: () => Promise; onDiscard: (commentId: string, revision: string) => Promise; } @@ -146,10 +146,11 @@ function SortableQueuedMessage({ {label} - {queue.targetRunId && (queue.protocol === "legacy" || entry.source?.requiresFreshSession ? ( + {(queue.targetRunId || queue.state === "deferred") && (!queue.targetRunId || queue.protocol === "legacy" || entry.source?.requiresFreshSession ? ( ) : (