diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 7e701455967..1319cdc26b7 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -29,6 +29,7 @@ import { issueRoutes } from "../routes/issues.js"; import { heartbeatService } from "../services/heartbeat.js"; import { issueService } from "../services/issues.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; +import { hasQueuedInteractionResponse } from "../services/queued-interaction-response.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"; @@ -668,6 +669,136 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running"); }); + // N-01: a redundant continuation wake (for example a recovery sweep that raced + // the response's own wake) carries a receipt the agent already received. It + // must not sit in the composer queue as an immutable, unremovable item. + const deliveredReceiptCases = [ + { kind: "ask_user_questions", status: "answered", title: null, label: "Answered: Questions", + payload: { version: 1, questions: [{ id: "goal", prompt: "What are you building?", selectionMode: "single", required: true, + options: [{ id: "shop", label: "A shop" }] }] }, result: { version: 1, answers: [{ questionId: "goal", optionIds: ["shop"] }] } }, + { kind: "request_confirmation", status: "accepted", title: "Approve hiring 2 agents", label: "Accepted: Approve hiring 2 agents", + payload: { version: 1, prompt: "Hire 2 agents?" }, result: { version: 1, outcome: "accepted" } }, + { kind: "request_confirmation", status: "rejected", title: "Approve hiring 2 agents", label: "Rejected: Approve hiring 2 agents", + payload: { version: 1, prompt: "Hire 2 agents?" }, result: { version: 1, outcome: "rejected" } }, + { kind: "request_checkbox_confirmation", status: "accepted", title: "Approve storefront prototype plan", + label: "Accepted: Approve storefront prototype plan", + payload: { version: 1, prompt: "Accept scope", options: [{ id: "scope", label: "Build the app" }] }, + result: { version: 1, outcome: "accepted", selectedOptionIds: ["scope"] } }, + { kind: "connection_intent", status: "accepted", title: "Connect GitHub", label: "Connected: GitHub connection", + payload: { version: 1, serviceName: "GitHub" }, result: { version: 1, outcome: "connected" } }, + { kind: "connection_intent", status: "rejected", title: "Connect Slack", label: "Declined: Slack connection", + payload: { version: 1, serviceName: "Slack" }, result: { version: 1, outcome: "declined" } }, + ] as const; + + async function seedDeliveredReceipt( + c: (typeof deliveredReceiptCases)[number], + delivering: "succeeded" | "running" | "failed" | "cancelled" | null, + runContext: "response" | "card_creation" = "response", + ) { + const seeded = await seedQueue(); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + const interactionId = randomUUID(); + const resolvedAt = new Date("2026-08-22T15:05:00.000Z"); + await db.insert(issueThreadInteractions).values({ + id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId, + kind: c.kind, status: c.status, createdByAgentId: seeded.agentId, sourceRunId: seeded.runId, + resolvedByUserId: "queue-owner", resolvedAt, + continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only", + effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit", + // Fixture payloads cover several interaction kinds; the schema types each kind separately. + title: c.title, payload: c.payload as never, result: c.result as never, + }); + const context = { issueId: seeded.issueId, interactionId, interactionKind: c.kind, interactionStatus: c.status }; + // The response's own wake already started a run that carries the answer. + const deliveringRunId = randomUUID(); + if (delivering) { + await db.insert(heartbeatRuns).values({ + id: deliveringRunId, companyId: seeded.companyId, agentId: seeded.agentId, + invocationSource: "automation", triggerDetail: "system", status: delivering, runtimeMode: "native", + startedAt: new Date("2026-08-22T15:05:01.000Z"), + finishedAt: delivering === "running" ? null : new Date("2026-08-22T15:06:00.000Z"), + // The card-creation wake carries the same interaction id while the + // card is still pending, without a resolved status. + contextSnapshot: runContext === "card_creation" + ? { issueId: seeded.issueId, interactionId, interactionKind: c.kind, wakeReason: "interaction_pending" } + : { ...context, source: "issue.interaction.respond", wakeReason: "issue_commented" }, + }); + } + // The redundant continuation, still deferred behind the issue's active run. + const staleWakeId = randomUUID(); + await db.insert(agentWakeupRequests).values({ + id: staleWakeId, companyId: seeded.companyId, agentId: seeded.agentId, + source: "automation", triggerDetail: "system", reason: "issue_continuation_needed", + status: "deferred_issue_execution", requestedByActorType: "system", requestedByActorId: "recovery", + requestedAt: new Date("2026-08-22T15:05:02.000Z"), + payload: { issueId: seeded.issueId, mutation: "interaction", interactionId, interactionStatus: c.status, + _paperclipWakeContext: { ...context, source: "issue.interaction_continuation_recovery" } }, + }); + return { ...seeded, interactionId, staleWakeId, deliveringRunId }; + } + + it.each(deliveredReceiptCases.flatMap(c => (["succeeded", "running"] as const).map(delivering => ({ ...c, delivering }))))( + "clears a $kind $status receipt once a $delivering run already delivered it", + async (c) => { + const seeded = await seedDeliveredReceipt(c, c.delivering); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary)).toEqual([]); + }, + ); + + it.each(deliveredReceiptCases)("keeps a pending $kind $status receipt until a run delivers it", async (c) => { + const seeded = await seedDeliveredReceipt(c, null); + const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries).toHaveLength(1); + expect(queue.body.entries[0]).toMatchObject({ + source: { kind: "interaction", interactionId: seeded.interactionId, interactionKind: c.kind, summary: c.label }, + canEdit: false, canDiscard: false, + }); + }); + + it.each([ + { delivering: "failed", runContext: "response" }, + { delivering: "cancelled", runContext: "response" }, + { delivering: "succeeded", runContext: "card_creation" }, + { delivering: "running", runContext: "card_creation" }, + ] as const)("keeps a receipt owed after a $delivering $runContext run", async ({ delivering, runContext }) => { + const seeded = await seedDeliveredReceipt(deliveredReceiptCases[0], delivering, runContext); + const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary)) + .toEqual(["Answered: Questions"]); + expect(await hasQueuedInteractionResponse(db, seeded.companyId, seeded.issueId, seeded.agentId)).toBe(true); + }); + + it("shows a later undelivered receipt instead of an earlier delivered one", async () => { + const seeded = await seedDeliveredReceipt(deliveredReceiptCases[3], "succeeded"); + const pendingId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: pendingId, companyId: seeded.companyId, issueId: seeded.issueId, + kind: "connection_intent", status: "rejected", createdByAgentId: seeded.agentId, + resolvedByUserId: "queue-owner", resolvedAt: new Date("2026-08-22T15:10:00.000Z"), + continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only", + effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit", + title: "Connect Slack", payload: { version: 1, serviceName: "Slack" } as never, result: { version: 1, outcome: "declined" } as never, + }); + await db.insert(agentWakeupRequests).values({ + companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system", + reason: "issue_commented", status: "deferred_issue_execution", requestedAt: new Date("2026-08-22T16:00:00.000Z"), + payload: { issueId: seeded.issueId, mutation: "interaction", interactionId: pendingId, interactionStatus: "rejected" }, + }); + const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary)) + .toEqual(["Declined: Slack connection"]); + }); + + it("does not hold agent handback for a receipt the agent already received", async () => { + const pending = await seedDeliveredReceipt(deliveredReceiptCases[1], null); + expect(await hasQueuedInteractionResponse(db, pending.companyId, pending.issueId, pending.agentId)).toBe(true); + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + const delivered = await seedDeliveredReceipt(deliveredReceiptCases[1], "succeeded"); + expect(await hasQueuedInteractionResponse(db, delivered.companyId, delivered.issueId, delivered.agentId)).toBe(false); + }); + it("promotes an approval after normal completion without a click", async () => { const seeded = await seedResponseQueue(); await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, seeded.runId)); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index e5b36b52793..852530786d1 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -42,7 +42,7 @@ import { resumeNativeWorkspaceAfterOwnerStop } from "../services/native-runtime/ import { retryNativeWorkspaceExport } from "../services/native-runtime/native-workspace-export-retry.js"; import { reconstructNativeWorkspace, unreconciledNativeExportPredicate } from "../services/native-runtime/native-workspace-reconstruction.js"; import { reconstructWorkspaceSchema } from "@paperclipai/shared"; -import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse } from "../services/queued-interaction-response.js"; +import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse, type QueuedInteractionResponse } from "../services/queued-interaction-response.js"; import { deliverConversationComments, isConversation } from "../services/agent-conversations.js"; import { assertOnboardingFirstTaskDependenciesAllowed } from "../services/onboarding-first-task-admission.js"; import { resolveChatTaskAssignment } from "../services/chat-task-assignment.js"; @@ -7403,6 +7403,8 @@ export function issueRoutes( wake: Pick; state: "deferred" | "queued"; queueRun: Pick | null; + /** The receipt already read for an interaction wake, reused by the queue builder. */ + response?: QueuedInteractionResponse | null; }; async function findQueuedCommentWake( @@ -7454,8 +7456,14 @@ export function issueRoutes( continue; const isReceipt = Boolean(queuedInteractionId(wake.payload)); if (isReceipt && receipt) continue; + // A receipt a run already received is not pending; a later one may be. + const readReceipt = () => readQueuedInteractionResponse(executor as Db, wake.companyId, issue.id, wake.payload); if (wake.status === "deferred_issue_execution") { - if (isReceipt) { receipt = { wake, state: "deferred", queueRun: null }; continue; } + if (isReceipt) { + const response = await readReceipt(); + if (response) receipt = { wake, state: "deferred", queueRun: null, response }; + continue; + } return { wake, state: "deferred", queueRun: null }; } if (!wake.runId) continue; @@ -7473,7 +7481,10 @@ export function issueRoutes( ) .limit(1) .then((runRows) => runRows[0] ?? null); - if (queueRun && isReceipt) receipt = { wake, state: "queued", queueRun }; + if (queueRun && isReceipt) { + const response = await readReceipt(); + if (response) receipt = { wake, state: "queued", queueRun, response }; + } else if (queueRun) return { wake, state: "queued", queueRun }; } return receipt; @@ -7483,11 +7494,9 @@ export function issueRoutes( executor: IssueQueueDb, issueId: string, wake: Pick | null, + response: QueuedInteractionResponse | null, ) { - if (wake && queuedInteractionId(wake.payload)) { - const response = await readQueuedInteractionResponse(executor as Db, wake.companyId, issueId, wake.payload); - return response ? [response.comment] : []; - } + if (wake && queuedInteractionId(wake.payload)) return response ? [response.comment] : []; const ids = queuedCommentIdsFromWakePayload(wake?.payload); if (ids.length === 0) return []; const rows = await executor @@ -7516,10 +7525,17 @@ export function issueRoutes( ? await findQueuedCommentWake(input.executor, input.issue) : input.queueState; const wake = queueState?.wake ?? null; + // Read an interaction receipt once per snapshot. + const response = wake && queuedInteractionId(wake.payload) + ? queueState?.response !== undefined + ? queueState.response + : await readQueuedInteractionResponse(input.executor as Db, input.issue.companyId, input.issue.id, wake.payload) + : null; const comments = await queueCommentsForWake( input.executor, input.issue.id, wake, + response, ); const assignedAgent = input.issue.assigneeAgentId ? await input.executor @@ -7563,7 +7579,6 @@ export function issueRoutes( actorId: input.actor.actorId, }); if (wake && queuedInteractionId(wake.payload)) { - const response = await readQueuedInteractionResponse(input.executor as Db, input.issue.companyId, input.issue.id, wake.payload); if (response?.source.requiresFreshSession) queue.steeringDisposition = "unsupported"; queue.entries = response ? [{ comment: response.comment, source: response.source, position: 0, canEdit: false, canDiscard: false }] : []; } diff --git a/server/src/services/queued-interaction-response.ts b/server/src/services/queued-interaction-response.ts index 553ed0efdaa..ff6959a7b95 100644 --- a/server/src/services/queued-interaction-response.ts +++ b/server/src/services/queued-interaction-response.ts @@ -34,15 +34,42 @@ export function queuedInteractionSummary(interaction: { return `${label}: ${subject}`; } +/** + * A run that started with this exact response in its context already received + * it. The run must carry the resolved status: the card-creation wake carries + * the same interaction id while the card is still pending, so it never counts. + * Any other wake that still carries the same receipt (for example a recovery + * continuation that raced the response's own wake) is redundant: it is not + * pending input, so it must not hold the composer queue or the issue. + * A cancelled or failed start does not count; that response is still owed. + */ +// Table-qualified on purpose: inside a select list Drizzle renders columns +// unqualified, which would bind them to the subquery's own run row. +function deliveredInteractionRunExists(companyId: string, issueId: string) { + return sql`EXISTS (SELECT 1 FROM heartbeat_runs delivered_run + WHERE delivered_run.company_id = ${companyId} + AND delivered_run.context_snapshot->>'issueId' = ${issueId} + AND delivered_run.context_snapshot->>'interactionId' = issue_thread_interactions.id::text + AND delivered_run.context_snapshot->>'interactionStatus' = issue_thread_interactions.status + AND delivered_run.started_at IS NOT NULL + AND delivered_run.status IN ('running', 'succeeded'))`; +} + +export type QueuedInteractionResponse = NonNullable>>; + export async function readQueuedInteractionResponse(db: Db, companyId: string, issueId: string, payload: unknown) { const id = queuedInteractionId(payload); if (!id) return null; - const [interaction] = await db.select().from(issueThreadInteractions).where(and( + const [row] = await db.select({ + interaction: issueThreadInteractions, + delivered: deliveredInteractionRunExists(companyId, issueId), + }).from(issueThreadInteractions).where(and( eq(issueThreadInteractions.id, id), eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId), inArray(issueThreadInteractions.status, ["accepted", "answered", "rejected"]), )); - if (!interaction || interaction.status !== record(payload).interactionStatus) return null; + const interaction = row?.interaction; + if (!interaction || interaction.status !== record(payload).interactionStatus || row.delivered) return null; const p = record(interaction.payload); const result = record(interaction.result); const heading = queuedInteractionSummary(interaction); @@ -80,6 +107,7 @@ export async function hasQueuedInteractionResponse(db: Db, companyId: string, is inArray(agentWakeupRequests.status, ["deferred_issue_execution", "queued"]), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`, + sql`NOT ${deliveredInteractionRunExists(companyId, issueId)}`, )).limit(1); return Boolean(row); }