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",