From f211d44c00fbc31d22e7ff687e22dfa05643da2a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 3 Nov 2026 22:09:58 -0600 Subject: [PATCH] fix(heartbeat): preserve queued coalesced comments at claim (#18913) ## Thinking Path > - Paperclip manages agents and their work. > - The wake queue records task comments before an agent starts. > - A queued run can accept another comment through coalescing. > - Claim rebuilds that run's input from its primary wake. > - Coalescing updated the run and its receipt but left the primary batch unchanged. > - This change saves the accepted batch atomically in the owning writer. > - The benefit is complete input at claim without repeating old work. ## Linked Issues or Issue Description Related: #18887 preserves offline queued turns. #13948 addresses equivalent automatic wake deduplication. This change repairs the accepted comment batch at the existing claim boundary. **What happened?** A queued run started with comment A. Admission accepted comment B, saved `[A, B]` in the run context, and finished B's coalesced receipt. The primary wake still held `[A]`. Actual claim rebuilt input from that wake and dropped B. **Expected behavior** Claim must retain every accepted live comment in order. Stop, changed source authority, deleted comments, and company boundaries must still apply. Already claimed wakes must retain their existing recovery source. **Steps to reproduce** 1. Queue a task run whose primary wake has an authoritative comment batch `[A]`. 2. Admit comment B through `createHeartbeatQueue.enqueueWakeup` and the real Postgres admission writer. 3. Confirm that admission accepts `[A, B]` and records B's coalesced receipt. 4. Claim the original queued run. Before this fix, the claimed input contains only A. **Paperclip version or commit** The untouched producer regression fails on `d7283e875772ca8c684dc86e471e6b48d543b7b1` with both process and native Runner/Codex adapter selection. Provider execution is disabled in the fixture. ## What Changed - Save the accepted comment batch in the unclaimed primary wake, run context, and coalesced receipt in one transaction. - Lock the primary wake before the run. Recheck the expected run status and same-company, same-agent and run bindings; apply the existing actor equality fence only to durable receipts before mutation. - Preserve incoming input as normal deferred work when the locked source changed. Preserve receipt identity and actor fields. - Preserve the original primary actor and trigger comment while updating its accepted batch. Each incoming receipt retains its actual actor. Keep claimed source payloads and legacy envelopes on their existing recovery paths. - Safely defer a same-named different agent without rewriting the original agent's run, wake or receipt. Missing and foreign-company targets remain rejected. Do not rebuild input from historical receipts. - Add actual process/native producer, real database rollback, Stop, deletion, foreign-source, and fallback controls. Update three direct writer fixtures for the required expected-status contract. ## Verification - The untouched process and native producer red on `d7283e875772ca8c684dc86e471e6b48d543b7b1` loses accepted comment B at actual claim. The initial incomplete mock is retained separately and excluded from behavioral proof. - Current process/native matrix: 22 passed, including mixed authors, system primary, same-named different agents, changed status, claimed sources, Stop, deletion and foreign-source boundaries. Queue, Postgres admission and use-case suites passed 234 cases before the source-preserving serializer follow-up; its 22 actual producer cases passed again afterward. - CI on the previous head found a real ordinary mixed-author workspace-copyback regression. The earlier 13-case selection omitted this test; that coverage gap is preserved. The first correction exposed a second real source-provenance/copyback failure. After preserving the primary trigger and actor while updating only its batch, the complete workspace recovery file passed all 188 cases, including ordinary mixed-author input and native copyback. - Real HTTP queued-comment consumption, discard/reorder and Stop boundaries: 101 passed before these review corrections. Existing chat and Slack consumers each passed their selected real writer case. - Previous-head CI also timed out in the CLI child-server startup fixture and a ZIP transfer case. The full untouched transfer file passed 17 cases locally. Startup remains unclassified; its intentionally reserved port is part of the fixture and is not diagnosed as the cause. No CI retry was issued. - An earlier queue run passed its 231 assertions but timed out during teardown; its preserved replay passed all 60 queue cases. This is distinct from the behavioral reds. - Exact `ab59d944895774d89f950248e356e93af00f4832` server type layers and production server/Runner build passed. The unchanged-default serializer file passed all 11 cases. Current-head CI and both substantive reviews remain required; previous grades and checks do not qualify this head. - No hosted task was retried or changed. Local tests do not establish any hosted deferred event's cause, provider execution, native acknowledgement, physical cleanup, or serving revision. ## Risks - Admission now performs bounded row reads and locks for each coalesced wake. It adds no idle scan or remote request. The existing task transaction owns the operation; lock order follows queued claim. - Claimed-wake recovery uses source identity. Its payload remains unchanged, and the existing actual native copyback tests cover that boundary. - A source that changes before the lock can defer the new input. This preserves work instead of consuming it into a stale run. No migration or historical backfill is required. ## Model Used OpenAI `gpt-6.1-sol`, with reasoning, repository inspection, shell execution, and test tools. The runtime does not report the context-window size. A separate OpenAI coordinator reviewed the source. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 17 +++ .../__tests__/chat-run-milestone-work.test.ts | 2 +- .../slack-conversation-lifecycle.test.ts | 2 +- .../wake-queue/adapters/postgres.test.ts | 34 ++++- .../modules/wake-queue/adapters/postgres.ts | 26 ++++ .../modules/wake-queue/application/ports.ts | 3 +- .../wake-queue/application/use-cases.test.ts | 16 +++ .../wake-queue/application/use-cases.ts | 6 +- server/src/services/heartbeat/queue.test.ts | 125 +++++++++++++++++- .../services/issue-queued-comment-queue.ts | 9 +- .../src/services/workspace-recovery.test.ts | 2 +- 11 files changed, 228 insertions(+), 14 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a5b8a5fe76f..2f22aac559f 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2766,6 +2766,23 @@ replay or access: the saved input and current principal, repository, conversatio runtime fence and execution containment still pass normal admission. Explicit conflicting App bindings remain denials. +When comments join a queued run with an unclaimed queued wake, admission saves +the accepted batch in both the run context and its authoritative primary wake in +the same transaction as the coalesced receipt. Claim rebuilds input from that primary wake and filters deleted +comments. It does not reconstruct input from historical coalesced receipts. +Admission locks and rechecks the run and its same-company, same-agent primary +wake. A changed status, mismatched binding, or mismatched durable-receipt actor preserves the incoming +input as deferred work instead of consuming it into the old source. Legacy wakes +without an authoritative batch retain their existing context-based claim path. +An already claimed wake keeps its source unchanged for its existing recovery +and replay path; this change does not grant it a new claim. +Ordinary mixed-author comments keep their existing coalescing behavior. The +primary wake retains its original actor, and each new receipt records its own +actual actor. Coalescing updates the queued batch while preserving the original +trigger comment fields, so recovery still verifies the original source actor. +Same-named agents with different IDs retain separate authority; +their new input is deferred without changing the other agent's queued work. + ### Queued input after a native Stop A run-only Stop ends the current response. It does not discard queued user diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 9f532ec8fc3..710ba505c68 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-work.test.ts @@ -330,7 +330,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.transaction(async tx => { await createWakeAdmissionWriter().coalesceIntoActiveExecutionRun( createAdmissionTransactionScope(f.companyId, tx as unknown as typeof db), { - companyId: f.companyId, activeExecutionRunId: seeded.id, mergedContextSnapshot: chatContextSnapshot(), + companyId: f.companyId, activeExecutionRunId: seeded.id, activeExecutionRunStatus: seeded.status, mergedContextSnapshot: chatContextSnapshot(), agentId: f.agentId, source: "on_demand", triggerDetail: null, payload: { issueId: f.issueId }, requestedByActorType: "user", requestedByActorId: f.userId, idempotencyKey: null, }); diff --git a/server/src/__tests__/slack-conversation-lifecycle.test.ts b/server/src/__tests__/slack-conversation-lifecycle.test.ts index 8644c13282b..6e2f5bf7031 100644 --- a/server/src/__tests__/slack-conversation-lifecycle.test.ts +++ b/server/src/__tests__/slack-conversation-lifecycle.test.ts @@ -206,7 +206,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.runId)); await db.transaction(async tx => { await createWakeAdmissionWriter().coalesceIntoActiveExecutionRun(createAdmissionTransactionScope(f.companyId, tx as unknown as typeof db), { - companyId: f.companyId, agentId: f.agentId, activeExecutionRunId: f.runId, + companyId: f.companyId, agentId: f.agentId, activeExecutionRunId: f.runId, activeExecutionRunStatus: "running", mergedContextSnapshot: { issueId: f.issueId }, source: "automation", triggerDetail: "system", payload: { issueId: f.issueId }, requestedByActorType: "system", requestedByActorId: null, idempotencyKey: key, diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index cfb8919cc80..3a8eae708c0 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -32,6 +32,7 @@ import { import type { WakeQueuePostgresAdapterDeps } from "./postgres.js"; import { createReleaseIssueExecution } from "../application/use-cases.js"; import type { TransactionScope } from "../application/ports.js"; +import { withQueuedCommentIdsInRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js"; // Proves the atomicity and company-scope properties the security review // requires: every mutation names `companyId` in its own SQL `WHERE` clause, @@ -1139,6 +1140,36 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { // owns it. These tests drive the admission writer directly against a // transaction they open themselves, the same way `heartbeat.ts` will. describe("wake admission", () => { + it.each(["running", "cancelled", "receipt-collision", "actor-mismatch"])("preserves both queued input sources after %s", async scenario => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const [oldComment] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user", authorUserId: "source-user", body: "Original input" }).returning(); + const [newComment] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user", authorUserId: "source-user", body: "Incoming input" }).returning(); + const runId = await seedRun({ companyId, agentId, status: ["receipt-collision", "actor-mismatch"].includes(scenario) ? "queued" : scenario, + contextSnapshot: withQueuedCommentIdsInRunContext({ issueId }, [oldComment.id]) }); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, runId, source: "assignment", status: "queued", + requestedByActorType: "user", requestedByActorId: "source-user", + payload: withQueuedCommentIdsInWakePayload({ issueId }, [oldComment.id]), + }).returning(); + await db.update(heartbeatRuns).set({ wakeupRequestId: wake.id }).where(eq(heartbeatRuns.id, runId)); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const operation = db.transaction(async tx => createWakeAdmissionWriter().coalesceIntoActiveExecutionRun( + createAdmissionTransactionScope(companyId, tx as unknown as Db), { + companyId, agentId, activeExecutionRunId: runId, activeExecutionRunStatus: "queued", + mergedContextSnapshot: withQueuedCommentIdsInRunContext({ issueId }, [oldComment.id, newComment.id]), + source: "on_demand", triggerDetail: "manual", payload: { issueId, commentId: newComment.id }, + requestedByActorType: "user", requestedByActorId: scenario === "actor-mismatch" ? "another-user" : "source-user", idempotencyKey: null, + ...(["receipt-collision", "actor-mismatch"].includes(scenario) + ? { durableReceipt: { id: scenario === "receipt-collision" ? wake.id : randomUUID(), requestedAt: new Date() } } : {}), + }, + )); + if (scenario === "receipt-collision") await expect(operation).rejects.toMatchObject({ cause: { code: "23505" } }); + else await expect(operation).resolves.toBeNull(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]).toEqual(before); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId))).toEqual([wake]); + }); + it("rolls back a deferred merge when its own durable receipt insert fails", async () => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); @@ -1241,7 +1272,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { try { await db.transaction(async tx => { await createWakeAdmissionWriter().coalesceIntoActiveExecutionRun(createAdmissionTransactionScope(companyId, tx as unknown as Db), { - companyId, activeExecutionRunId: runId, mergedContextSnapshot: { commentId: "fresh-input" }, + companyId, activeExecutionRunId: runId, activeExecutionRunStatus: "running", mergedContextSnapshot: { commentId: "fresh-input" }, agentId, source: "on_demand", triggerDetail: null, payload: null, requestedByActorType: "user", requestedByActorId: null, idempotencyKey: null, }); @@ -1269,6 +1300,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { await writer.coalesceIntoActiveExecutionRun(scope, { companyId: otherCompanyId, activeExecutionRunId: runId, + activeExecutionRunStatus: "running", mergedContextSnapshot: { taskKey: "issue-1", commentId: "c1" }, agentId, source: "on_demand", diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 5c328e25bce..8cfe44af184 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -49,6 +49,7 @@ import { withRecoveryContext } from "../../../services/recovery/status-only-cont import { parseIssueExecutionState } from "../../../services/issue-execution-policy.js"; import { queuedCommentIdsFromWakePayload, + queuedCommentIdsFromRunContext, withQueuedCommentIdsInWakePayload, } from "../../../services/issue-queued-comment-queue.js"; import { CHAT_CONVERSATION_ID_KEY, extractWakeCommentIds } from "../../run-dispatch/index.js"; @@ -971,7 +972,32 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { async coalesceIntoActiveExecutionRun(scope, input) { const tx = requireAdmissionTx(scope, input.companyId); const now = new Date(); + const [candidate] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.activeExecutionRunId), eq(heartbeatRuns.companyId, input.companyId), + )).limit(1); + if (!candidate) throw new Error("wake-queue: the coalesce target run was not found for this company"); + if (candidate.agentId !== input.agentId) return null; + const [primary] = candidate.wakeupRequestId ? await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, candidate.wakeupRequestId), eq(agentWakeupRequests.companyId, input.companyId), + eq(agentWakeupRequests.agentId, input.agentId), + )).for("update").limit(1) : []; + const [current] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, candidate.id), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.agentId, input.agentId), + )).for("update").limit(1); + if (!current || current.status !== input.activeExecutionRunStatus || current.wakeupRequestId !== candidate.wakeupRequestId) return null; + const commentIds = queuedCommentIdsFromRunContext(input.mergedContextSnapshot); + const synchronizeBatch = current.status === "queued" && commentIds.length > 0 + && primary?.status === "queued" && queuedCommentIdsFromWakePayload(primary.payload).length > 0; + if (current.status === "queued" && commentIds.length && current.wakeupRequestId && + (!primary || !["queued", "claimed"].includes(primary.status) || primary.runId !== current.id + || (input.durableReceipt && (primary.requestedByActorType !== input.requestedByActorType || primary.requestedByActorId !== input.requestedByActorId)) + || primary.payload?.issueId !== input.mergedContextSnapshot.issueId)) return null; await notifyChatRunMilestones(tx, { companyId: input.companyId, runId: input.activeExecutionRunId }); + if (synchronizeBatch && primary) await tx.update(agentWakeupRequests).set({ + payload: withQueuedCommentIdsInWakePayload(primary.payload, commentIds, { preserveSource: true }), updatedAt: now, + }).where(and(eq(agentWakeupRequests.id, primary.id), eq(agentWakeupRequests.companyId, input.companyId), + eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.runId, current.id), eq(agentWakeupRequests.status, primary.status))); const mergedRun = await tx .update(heartbeatRuns) .set({ contextSnapshot: input.mergedContextSnapshot, updatedAt: now }) diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index 66c8a631c15..500d1675bc2 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -425,6 +425,7 @@ export interface WakeAdmissionWriter { input: { companyId: string; activeExecutionRunId: string; + activeExecutionRunStatus: WakeAdmissionActiveExecutionRun["status"]; mergedContextSnapshot: Record; durableReceipt?: DurableWakeAdmissionReceipt; agentId: string; @@ -435,7 +436,7 @@ export interface WakeAdmissionWriter { requestedByActorId: string | null; idempotencyKey: string | null; }, - ): Promise>; + ): Promise | null>; /** Merges the wake's context into an already-queued deferred wake, guarded by its current status. */ mergeIntoExistingDeferredWake( scope: TransactionScope, diff --git a/server/src/modules/wake-queue/application/use-cases.test.ts b/server/src/modules/wake-queue/application/use-cases.test.ts index 7f605e70f30..0677818c9c2 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -1228,6 +1228,22 @@ describe("admitWakeBehindIssueExecution", () => { expect(reader.findExistingDeferredWake).not.toHaveBeenCalled(); }); + it("retains incoming receipt authority when the locked coalesce target has changed", async () => { + const writer = createFakeAdmissionWriter({ coalesceIntoActiveExecutionRun: vi.fn(async () => null) }); + const reader = createFakeAdmissionReader(); + const admit = createAdmitWakeBehindIssueExecution({ reader, writer, helpers: createFakeAdmissionHelpers() }); + const durableReceipt = { id: "incoming-source", requestedAt: new Date("2026-11-04T00:00:00Z") }; + expect(await admit(SCOPE, admissionInput({ durableReceipt, wakeCommentId: "new-comment", + contextSnapshot: { issueId: "issue-1", wakeCommentId: "new-comment" }, payload: { issueId: "issue-1", commentId: "new-comment" }, + }))).toEqual({ kind: "deferred" }); + expect(writer.insertNewDeferredWake).toHaveBeenCalledWith(SCOPE, expect.objectContaining({ + companyId: "company-1", agentId: "wake-agent", durableReceipt, + requestedByActorType: "user", requestedByActorId: "user-1", + payload: expect.objectContaining({ issueId: "issue-1", commentId: "new-comment" }), + })); + expect(writer.mergeIntoExistingDeferredWake).not.toHaveBeenCalled(); + }); + it("merges into the existing deferred wake when the policy returns a merge target", async () => { const mergeIntoExistingDeferredWake = vi.fn( async (_scope: TransactionScope, _input: Parameters[1]) => {}, diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 79aada9b05f..6484ee18b9a 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -870,6 +870,7 @@ export function createAdmitWakeBehindIssueExecution(deps: { const run = await deps.writer.coalesceIntoActiveExecutionRun(scope, { companyId: input.companyId, activeExecutionRunId: target.id, + activeExecutionRunStatus: target.status, mergedContextSnapshot, ...(input.durableReceipt ? { durableReceipt: input.durableReceipt } @@ -882,12 +883,9 @@ export function createAdmitWakeBehindIssueExecution(deps: { requestedByActorId: input.requestedByActorId, idempotencyKey: input.idempotencyKey, }); - return { kind: "coalesced", run }; + if (run) return { kind: "coalesced", run }; } - // decision.kind === "defer": only now does the module read for an - // existing deferred wake, so the coalesce path (the common path) never - // pays for this query. const existingDeferred = (interactionReceipt || input.allowRunCoalescing === false) ? null diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index ef92bf94c65..1dd39dd5067 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -1,13 +1,16 @@ import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js"; import type { Db } from "@paperclipai/db"; -import { authUsers, agents, agentWakeupRequests, caseHandoffs, cases, chatCompletionDeliveries, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issueThreadInteractions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { authUsers, agents, agentWakeupRequests, caseHandoffs, cases, chatCompletionDeliveries, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issueThreadInteractions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; -import { createHeartbeatQueue, type HeartbeatQueueDependencies } from "./queue.js"; +import { createHeartbeatQueue, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake, type HeartbeatQueueDependencies } from "./queue.js"; +import { createWakeQueue } from "../../modules/wake-queue/index.js"; +import { mergeCoalescedContextSnapshot } from "./run-preparation.js"; +import { withQueuedCommentIdsInWakePayload, withQueuedCommentIdsInRunContext } from "../issue-queued-comment-queue.js"; import { settleConversationTurn } from "../agent-conversations.js"; import { chatCompletionDeliveryService, recordChatCompletion, recordChatHandoff } from "../chat-completion-delivery.js"; import { instanceSettingsService } from "../instance-settings.js"; @@ -256,6 +259,124 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { return { company, agent, issue, wake, run }; } + it.each((["process", "paperclip_runner"] as const).flatMap(adapterType => (["coalesced", "claimed-primary", "legacy-envelope", "delete-old", "delete-new", "stopped", "running", "system-primary", "foreign-primary", "same-named-agent", "mixed-author"] as const).map(scenario => ({ adapterType, scenario }))))("protects a queued comment batch across $adapterType/$scenario", async ({ adapterType, scenario }) => { + const f = await fixture("queued"); + let admissionAgent: typeof agents.$inferSelect = { ...f.agent, adapterType, adapterConfig: adapterType === "paperclip_runner" ? { provider: "codex" } : {} }; + await db.update(agents).set({ adapterType, adapterConfig: admissionAgent.adapterConfig }).where(eq(agents.id, f.agent.id)); + if (scenario === "same-named-agent") [admissionAgent] = await db.insert(agents).values({ companyId: f.company.id, + name: f.agent.name.toUpperCase(), role: "engineer", status: "idle", adapterType, adapterConfig: admissionAgent.adapterConfig, + }).returning(); + const userId = `source-user-${f.company.id}`; + await db.insert(authUsers).values({ id: userId, name: "Source owner", email: `${f.company.id}@queue.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId: f.company.id, principalType: "user", principalId: userId, status: "active", membershipRole: "operator" }); + await db.update(companies).set({ defaultResponsibleUserId: userId }).where(eq(companies.id, f.company.id)); + const incomingActorId = scenario === "mixed-author" ? `second-user-${f.company.id}` : userId; + if (scenario === "mixed-author") { + await db.insert(authUsers).values({ id: incomingActorId, name: "Incoming source owner", email: `${incomingActorId}@queue.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId: f.company.id, principalType: "user", principalId: incomingActorId, status: "active", membershipRole: "operator" }); + } + const [oldComment] = await db.insert(issueComments).values({ companyId: f.company.id, issueId: f.issue.id, + authorType: "user", authorUserId: userId, body: "Review source A" }).returning(); + const [newComment] = await db.insert(issueComments).values({ companyId: f.company.id, issueId: f.issue.id, + authorType: "user", authorUserId: incomingActorId, body: "Review source B" }).returning(); + await db.update(agentWakeupRequests).set({ status: scenario === "claimed-primary" ? "claimed" : "queued", runId: f.run.id, claimedAt: null, + requestedByActorType: scenario === "system-primary" ? "system" : "user", requestedByActorId: scenario === "system-primary" ? null : userId, + payload: scenario === "legacy-envelope" ? { issueId: f.issue.id, commentId: oldComment.id } + : withQueuedCommentIdsInWakePayload({ issueId: f.issue.id, retainedSource: "original-source" }, [oldComment.id]), + }).where(eq(agentWakeupRequests.id, f.wake.id)); + const [scanned] = await db.update(heartbeatRuns).set({ responsibleUserId: userId, + contextSnapshot: withQueuedCommentIdsInRunContext({ issueId: f.issue.id, wakeReason: "issue_commented" }, [oldComment.id]), + }).where(eq(heartbeatRuns.id, f.run.id)).returning(); + let foreignPrimary: typeof agentWakeupRequests.$inferSelect | undefined; + if (scenario === "foreign-primary") { + const [company] = await db.insert(companies).values({ name: "Foreign source", issuePrefix: "FOREIGN" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Foreign source agent", adapterType: "process" }).returning(); + [foreignPrimary] = await db.insert(agentWakeupRequests).values({ companyId: company.id, agentId: agent.id, source: "assignment", status: "queued", payload: { source: "foreign" } }).returning(); + await db.update(heartbeatRuns).set({ wakeupRequestId: foreignPrimary.id }).where(eq(heartbeatRuns.id, f.run.id)); + } + if (scenario === "delete-old") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, oldComment.id)); + const deps = callbacks(db); + deps.getAgent.mockResolvedValue(admissionAgent); + deps.canStartNewExternalBotWork.mockResolvedValue(false); + deps.instanceSettings.getExperimental.mockResolvedValue({ enableIsolatedWorkspaces: false, enableNativeRunner: true } as Awaited>); + const admission = createWakeQueue(db, { + resolveResponsibleUserId: async () => userId, + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), + resolveSessionBeforeForWakeup: async () => null, + wakeAdmissionHelpers: { filterZombieCoalesceTarget: deps.filterZombieCoalesceTarget, + mergeCoalescedContextSnapshot, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake }, + recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined }, + }); + deps.wakeQueue.createAdmissionTransactionScope.mockImplementation(admission.createAdmissionTransactionScope); + deps.wakeQueue.admitWakeBehindIssueExecution.mockImplementation(admission.admitWakeBehindIssueExecution); + const queue = createHeartbeatQueue(db, deps); + if (scenario === "running") { + deps.canStartNewExternalBotWork.mockResolvedValue(true); + expect(await queue.claimQueuedRun(scanned)).toMatchObject({ status: "running" }); + deps.canStartNewExternalBotWork.mockResolvedValue(false); + } + const [originalPrimary] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wake.id)); + const coalesced = await queue.enqueueWakeup(admissionAgent.id, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + payload: { issueId: f.issue.id, commentId: newComment.id }, contextSnapshot: { issueId: f.issue.id, wakeCommentId: newComment.id }, + requestedByActorType: "user", requestedByActorId: incomingActorId }); + if (["running", "foreign-primary", "same-named-agent"].includes(scenario)) { + expect(coalesced).toBeNull(); + const [preserved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)); + expect(preserved.contextSnapshot).toMatchObject({ wakeCommentId: oldComment.id, wakeCommentIds: [oldComment.id] }); + const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.company.id)); + expect(receipts.find(row => row.id !== f.wake.id)).toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(receipts.find(row => row.id === f.wake.id)).toEqual(originalPrimary); + if (scenario === "same-named-agent") { + expect(preserved).toEqual(scanned); + expect(receipts.find(row => row.id !== f.wake.id)).toMatchObject({ agentId: admissionAgent.id, + requestedByActorType: "user", requestedByActorId: incomingActorId, payload: { issueId: f.issue.id, commentId: newComment.id } }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.company.id))).toEqual([scanned]); + } + if (foreignPrimary) expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, foreignPrimary.id)))[0]).toEqual(foreignPrimary); + expect(deps.executeRun).not.toHaveBeenCalled(); + return; + } + expect(coalesced).toMatchObject({ id: f.run.id, status: "queued", contextSnapshot: { wakeCommentIds: [oldComment.id, newComment.id] } }); + const accepted = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.company.id)); + expect(accepted.find(row => row.id !== f.wake.id)).toMatchObject({ status: "coalesced", runId: f.run.id, finishedAt: expect.any(Date), + requestedByActorType: "user", requestedByActorId: incomingActorId }); + const [updatedPrimary] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wake.id)); + expect(updatedPrimary).toMatchObject({ id: originalPrimary.id, companyId: originalPrimary.companyId, + agentId: originalPrimary.agentId, source: originalPrimary.source, requestedByActorType: originalPrimary.requestedByActorType, + requestedByActorId: originalPrimary.requestedByActorId, idempotencyKey: originalPrimary.idempotencyKey, requestedAt: originalPrimary.requestedAt }); + if (scenario === "legacy-envelope" || scenario === "claimed-primary") expect(updatedPrimary.payload).toEqual(originalPrimary.payload); + else expect(updatedPrimary.payload).toMatchObject({ retainedSource: "original-source", commentId: oldComment.id, + _paperclipWakeContext: { commentId: oldComment.id, wakeCommentId: oldComment.id, wakeCommentIds: [oldComment.id, newComment.id] } }); + if (scenario === "claimed-primary") { + deps.canStartNewExternalBotWork.mockResolvedValue(true); + expect(await queue.claimQueuedRun(scanned)).toBeNull(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]) + .toMatchObject({ status: "queued", contextSnapshot: { wakeCommentIds: [oldComment.id, newComment.id] } }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wake.id)))[0]).toEqual(originalPrimary); + expect(deps.executeRun).not.toHaveBeenCalled(); + return; + } + if (scenario === "stopped") { + await createHeartbeatRunControl(db, runControlCallbacks(db)).cancelRunInternal(f.run.id, "operator Stop"); + deps.canStartNewExternalBotWork.mockResolvedValue(true); + expect(await queue.claimQueuedRun(scanned)).toBeNull(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0]) + .toMatchObject({ status: "cancelled", contextSnapshot: { wakeCommentIds: [oldComment.id, newComment.id] } }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wake.id)))[0].status).toBe("cancelled"); + expect(deps.executeRun).not.toHaveBeenCalled(); + return; + } + if (scenario === "delete-new") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, newComment.id)); + deps.canStartNewExternalBotWork.mockResolvedValue(true); + const claimed = await queue.claimQueuedRun(scanned); + expect(claimed).toMatchObject({ id: f.run.id, status: "running", + contextSnapshot: { wakeCommentId: scenario === "delete-new" ? oldComment.id : newComment.id, + wakeCommentIds: scenario === "delete-old" ? [newComment.id] : scenario === "delete-new" ? [oldComment.id] : [oldComment.id, newComment.id] } }); + const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.company.id)); + expect(receipts.find(row => row.id !== f.wake.id)).toMatchObject({ status: "coalesced", runId: f.run.id }); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + async function strandedMention(runtimeMode = "legacy") { const seeded = await fixture("succeeded"); await db.update(heartbeatRuns).set({ runtimeMode, nativeIssueId: runtimeMode === "native" ? seeded.issue.id : null, finishedAt: new Date() }).where(eq(heartbeatRuns.id, seeded.run.id)); diff --git a/server/src/services/issue-queued-comment-queue.ts b/server/src/services/issue-queued-comment-queue.ts index 60ca8efea1e..e50809fe6c7 100644 --- a/server/src/services/issue-queued-comment-queue.ts +++ b/server/src/services/issue-queued-comment-queue.ts @@ -33,15 +33,18 @@ export function queuedCommentIdsFromRunContext(contextValue: unknown): string[] export function withQueuedCommentIdsInWakePayload( payloadValue: unknown, ids: string[], + options: { preserveSource?: boolean } = {}, ): Record { const payload = { ...record(payloadValue) }; const context = { ...record(payload[QUEUE_CONTEXT_KEY]) }; if (ids.length > 0) { const latestId = ids[ids.length - 1]!; context[QUEUE_IDS_KEY] = ids; - context.wakeCommentId = latestId; - context.commentId = latestId; - payload.commentId = latestId; + if (!options.preserveSource) { + context.wakeCommentId = latestId; + context.commentId = latestId; + payload.commentId = latestId; + } } else { delete context[QUEUE_IDS_KEY]; delete context.wakeCommentId; diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index b2d4f1c9155..f6b2177ab78 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -2433,7 +2433,7 @@ describe("automatic retained workspace recovery", () => { expect(await createWakeAdmissionReader().matchesActiveWakeActor(scope, actor)).toBe(true); expect(await createWakeAdmissionReader().matchesActiveWakeActor(scope, { ...actor, requestedByActorId: "other-user" })).toBe(false); await createWakeAdmissionWriter().coalesceIntoActiveExecutionRun(scope, { - companyId: f.companyId, agentId: original.agentId, activeExecutionRunId: queued.id, + companyId: f.companyId, agentId: original.agentId, activeExecutionRunId: queued.id, activeExecutionRunStatus: queued.status, mergedContextSnapshot: mergeCoalescedContextSnapshot((await run(queued.id)).contextSnapshot, incoming, { preserveExistingInteractionContinuation: true }), source: "on_demand", triggerDetail: "manual",