diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 4f83a32fce5..e329c8c50f5 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -170,6 +170,13 @@ to existing answered threads without another model invocation. Slack identity, permissions, and the ability to execute work in the same thread are unchanged. Other providers keep their existing lifecycle. +Several chat conversations can share one task, such as the inline finding +threads of a GitHub pull request review. A reviewed chat run attests exactly one +conversation, so wake coalescing keeps each conversation in its own batch. +Messages from the same conversation still coalesce into one run. Messages from +different conversations queue separately and run one after the other. Wakes +without a conversation keep their normal coalescing. + ### `done` The work is complete and terminal. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index fe72a7e49f9..7d3c18dadd4 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -184,12 +184,14 @@ import { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON, } from "../services/chat-run-publications.js"; import { + extractWakeCommentIds, heartbeatService, resolveExternalChatWakeProvider, } from "../services/heartbeat.js"; import { buildPaperclipWakePayload, resolveRunScopedMentionedSkillKeys } from "../services/heartbeat/run-preparation.js"; import { registerServerAdapter, + runningProcesses, unregisterServerAdapter, } from "../adapters/index.js"; import { projectSafeChatPublicationText } from "../services/chat-publication-projection.js"; @@ -3158,6 +3160,103 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(await githubBotToolsForSession(db, { ...f.session, issueId: randomUUID() })).toEqual([]); expect(await readChatConversationsByEndpointId(f.endpoint.id)).toHaveLength(2); }); + describe("replies in review finding threads behind an active run", () => { + const pr = "github:paperclipai/paperclip:17044"; + async function findingThreadFixture() { + vi.stubEnv("PAPERCLIP_API_URL", "https://paperclip.example.test"); + onTestFinished(() => vi.unstubAllEnvs()); + const f = await subscribedGitHubFixture(pr); + const adapterType = `github-finding-thread-test-${randomUUID()}`; + const execute = vi.fn(async (_input: { runId: string; context: Record }) => ({ + exitCode: 0, signal: null, timedOut: false, errorMessage: null, summary: "Answered the finding thread", + })); + registerServerAdapter({ type: adapterType, supportsLocalAgentJwt: false, execute, + testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }) }); + const heartbeat = heartbeatService(db); + const activeRunId = randomUUID(); + onTestFinished(async () => { + runningProcesses.delete(activeRunId); + await heartbeat.drainActiveRunExecutions(); + unregisterServerAdapter(adapterType); + }); + const issueId = f.conversation.issueId!; + await db.update(companies).set({ defaultResponsibleUserId: "owner-user" }).where(eq(companies.id, f.companyId)); + await db.update(agents).set({ adapterType, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }) + .where(eq(agents.id, f.assignedAgentId)); + // Published findings bind their inline threads to the PR review task. + const { id: _id, ...base } = f.conversation; + const threads = await Promise.all(["4332012267", "4332012626"].map(async (commentId) => { + const thread = makeThread({ channelId: "paperclipai/paperclip", id: `${pr}:rc:${commentId}` }).thread; + const [conversation] = await db.insert(chatConversations).values({ ...base, externalThreadId: thread.id, + providerUrl: `https://github.com/paperclipai/paperclip/pull/17044#discussion_r${commentId}` }).returning(); + return { thread, conversation }; + })); + // The review-request run still holds the task when the linked user replies. + await db.insert(heartbeatRuns).values({ id: activeRunId, companyId: f.companyId, agentId: f.assignedAgentId, + status: "running", startedAt: new Date(), contextSnapshot: { issueId, source: "chat:github" }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }); + runningProcesses.set(activeRunId, { child: {} as never, graceSec: 0, processGroupId: null }); + await db.update(issues).set({ status: "in_review", executionRunId: activeRunId, + executionAgentNameKey: "maya", executionLockedAt: new Date() }).where(eq(issues.id, issueId)); + f.wakeup.mockImplementation(heartbeat.wakeup); + let messageId = 41950; + const reply = (thread: Thread, text: string) => deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, + provider: "github", thread, trigger: "subscribed_message", + message: makeMessage({ id: String(messageId++), text, userId: "42", userName: "octocat" }) }); + const commentFor = async (text: string) => (await readIssueCommentsByIssueId(issueId)).find((c) => c.body.includes(text))!.id; + const finishActiveRun = async () => { + runningProcesses.delete(activeRunId); + await heartbeat.cancelRun(activeRunId, "Review request finished", { suppressImmediateRecovery: true }); + await heartbeat.drainActiveRunExecutions(); + }; + const followUpRuns = async () => (await readHeartbeatRunsByCompanyId(f.companyId)) + .filter((run) => run.id !== activeRunId && run.wakeupRequestId) + .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()); + return { ...f, issueId, threads, execute, reply, commentFor, finishActiveRun, followUpRuns }; + } + + it("answers each finding thread in its own run instead of one unattestable batch", async () => { + const f = await findingThreadFixture(); + await f.reply(f.threads[0].thread, "First finding is intentional"); + await f.reply(f.threads[1].thread, "Second finding is fixed in the next commit"); + const first = await f.commentFor("First finding"), second = await f.commentFor("Second finding"); + await f.finishActiveRun(); + const runs = await f.followUpRuns(); + expect(runs.map((run) => ({ status: run.status, errorCode: run.errorCode, error: run.error }))) + .toEqual([{ status: "succeeded", errorCode: null, error: null }, { status: "succeeded", errorCode: null, error: null }]); + expect(runs.map((run) => extractWakeCommentIds(run.contextSnapshot))).toEqual([[first], [second]]); + expect(runs.map((run) => run.contextSnapshot?.chatConversationId)) + .toEqual(f.threads.map(({ conversation }) => conversation.id)); + expect(f.execute.mock.calls.map(([input]) => [input.context.paperclipExternalChatExecutionBound, + extractWakeCommentIds(input.context)])).toEqual([[true, [first]], [true, [second]]]); + }); + + it("still batches several replies in one finding thread into one attested run", async () => { + const f = await findingThreadFixture(); + await f.reply(f.threads[0].thread, "First note on this finding"); + await f.reply(f.threads[0].thread, "Second note on this finding"); + const ids = [await f.commentFor("First note"), await f.commentFor("Second note")]; + await f.finishActiveRun(); + const runs = await f.followUpRuns(); + expect(runs).toHaveLength(1); + expect(runs[0]).toMatchObject({ status: "succeeded", errorCode: null }); + expect(extractWakeCommentIds(runs[0].contextSnapshot)).toEqual(ids); + expect(f.execute).toHaveBeenCalledTimes(1); + }); + + it("still denies each separated finding-thread run after the linked identity is revoked", async () => { + const f = await findingThreadFixture(); + await f.reply(f.threads[0].thread, "First finding is intentional"); + await f.reply(f.threads[1].thread, "Second finding is fixed in the next commit"); + await db.update(chatIdentityLinks).set({ status: "revoked", revokedAt: new Date() }) + .where(eq(chatIdentityLinks.principalId, f.principal.id)); + await f.finishActiveRun(); + const runs = await f.followUpRuns(); + expect(runs.length).toBeGreaterThan(0); + for (const run of runs) expect(run).toMatchObject({ status: "failed", errorCode: "setup_failed" }); + expect(f.execute).not.toHaveBeenCalled(); + }); + }); it.each([ { label: "returns the confirmed GitHub assessment when publication takes longer than ten seconds", delayMs: 11_000, timeoutMs: undefined }, { label: "preserves an explicit shorter GitHub tool deadline and reconciles without duplicate publication", delayMs: 200, timeoutMs: 50 }, diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index a25833758f0..fd944751bfe 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -75,6 +75,25 @@ export function deriveCommentId( ); } +/** + * The external chat conversation a wake answers. Several conversations can + * share one issue (for example GitHub review finding threads), but a reviewed + * chat run can attest only one, so wakes for different conversations must + * never share a batch. + */ +export const CHAT_CONVERSATION_ID_KEY = "chatConversationId"; + +export function readWakeChatConversationId(contextSnapshot: unknown): string | null { + return readNonEmptyString(parseObject(contextSnapshot)[CHAT_CONVERSATION_ID_KEY]); +} + +/** True when both wakes answer a chat conversation and they are different ones. */ +export function wakeChatConversationsConflict(existing: unknown, incoming: unknown): boolean { + const existingId = readWakeChatConversationId(existing); + const incomingId = readWakeChatConversationId(incoming); + return existingId !== null && incomingId !== null && existingId !== incomingId; +} + /** * `allowedWakeReasons` is the issue-tree-control module's own set of wake * reasons that excuse an interaction wake. This file stays free of service diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts index d29b0783231..5293ca0ae64 100644 --- a/server/src/modules/run-dispatch/index.ts +++ b/server/src/modules/run-dispatch/index.ts @@ -17,6 +17,9 @@ export { INTERACTION_CONTINUATION_INFRA_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_WAKE_REASON, WAKE_COMMENT_IDS_KEY, + CHAT_CONVERSATION_ID_KEY, + readWakeChatConversationId, + wakeChatConversationsConflict, RESOLVED_INTERACTION_CONTINUATION_STATUSES, isNonAssigneeWorkspaceBusyRetry, extractWakeCommentIds, diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index d147c3042f5..a74aaf3ecc6 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -40,7 +40,7 @@ import { queuedCommentIdsFromWakePayload, withQueuedCommentIdsInWakePayload, } from "../../../services/issue-queued-comment-queue.js"; -import { extractWakeCommentIds } from "../../run-dispatch/index.js"; +import { CHAT_CONVERSATION_ID_KEY, extractWakeCommentIds } from "../../run-dispatch/index.js"; import { hasInteractionContinuationWakeContext } from "../domain/context.js"; import { decidePreDrain, type PreDrainFacts } from "../domain/policy.js"; import { @@ -890,7 +890,7 @@ export function createWakeAdmissionReader(): WakeAdmissionReader { async findExistingDeferredWake( scope, - { companyId, agentId, issueId, durableActor }, + { companyId, agentId, issueId, chatConversationId, durableActor }, ) { const tx = requireAdmissionTx(scope, companyId); const row = await tx @@ -902,6 +902,11 @@ export function createWakeAdmissionReader(): WakeAdmissionReader { eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, + ...(chatConversationId + ? [ + sql`coalesce(${agentWakeupRequests.payload} -> ${DEFERRED_WAKE_CONTEXT_KEY}::text ->> ${CHAT_CONVERSATION_ID_KEY}::text, ${chatConversationId}) = ${chatConversationId}`, + ] + : []), ...(durableActor ? [ durableActor.type === null diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index 6180288d797..c37aa995e0e 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -407,6 +407,8 @@ export interface WakeAdmissionReader { companyId: string; agentId: string; issueId: string; + /** Matches only deferred wakes for this chat conversation or for no conversation. */ + chatConversationId?: string | null; durableActor?: { type: string | null; id: string | null }; }, ): Promise; 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 c05248a85cb..0c7e66f9dd1 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -937,6 +937,29 @@ describe("admitWakeBehindIssueExecution", () => { expect(writer.insertNewDeferredWake).toHaveBeenCalledOnce(); }); + it.each([ + { label: "another conversation's queued run", active: "conversation-a", incoming: "conversation-b", coalesces: false }, + { label: "the same conversation's queued run", active: "conversation-a", incoming: "conversation-a", coalesces: true }, + { label: "a queued run without a conversation", active: undefined, incoming: "conversation-b", coalesces: true }, + { label: "a conversation's queued run from a non-chat wake", active: "conversation-a", incoming: undefined, coalesces: true }, + ])("coalesces a chat reply into $label only when one conversation is attested", async ({ active, incoming, coalesces }) => { + const writer = createFakeAdmissionWriter(); + const reader = createFakeAdmissionReader(); + const admit = createAdmitWakeBehindIssueExecution({ reader, writer, helpers: createFakeAdmissionHelpers() }); + await admit(SCOPE, admissionInput({ + contextSnapshot: incoming ? { chatConversationId: incoming } : {}, + activeExecutionRun: { ...ACTIVE_EXECUTION_RUN, status: "queued", + contextSnapshot: active ? { chatConversationId: active } : {} }, + })); + expect(writer.coalesceIntoActiveExecutionRun).toHaveBeenCalledTimes(coalesces ? 1 : 0); + if (!coalesces) { + // The separated reply still merges with its own conversation's deferred wake. + expect(reader.findExistingDeferredWake).toHaveBeenCalledExactlyOnceWith(SCOPE, + expect.objectContaining({ chatConversationId: incoming })); + expect(writer.insertNewDeferredWake).toHaveBeenCalledOnce(); + } + }); + it("partitions durable admission by the exact actor before considering a deferred merge", async () => { const durableReceipt = { id: "other-actor-receipt", @@ -983,6 +1006,7 @@ describe("admitWakeBehindIssueExecution", () => { companyId: "company-1", agentId: "wake-agent", issueId: "issue-1", + chatConversationId: null, durableActor: { type: "user", id: "other-user" }, }, ); diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index d1b0b326f45..1e3ffa93609 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -1,4 +1,5 @@ -import { enrichPromotedWakeContext } from "../domain/context.js"; +import { readWakeChatConversationId } from "../../run-dispatch/index.js"; +import { enrichPromotedWakeContext, wakeMustNotJoinActiveRun } from "../domain/context.js"; import { decideQueuedCommentAction, decideReleaseRecovery, @@ -11,7 +12,6 @@ import { isConfigurationIncompleteFailedRun, isWorkspaceValidationFailedRun, readNonEmptyString, - parseObject, } from "../domain/values.js"; import type { AdmitWakeBehindIssueExecutionResult, @@ -840,11 +840,10 @@ export function createAdmitWakeBehindIssueExecution(deps: { requestedByActorType: input.requestedByActorType, requestedByActorId: input.requestedByActorId, })); - // Each resolved card has its own immutable decision and continuation. Never - // merge it with comments or another card, in either arrival order. + // A resolved card never merges with comments or another card. const interactionReceipt = Boolean(input.contextSnapshot.interactionId || input.payload?.interactionId); const decision = decideWakeAdmission({ - allowRunCoalescing: interactionReceipt || parseObject(input.activeExecutionRun.contextSnapshot).interactionId + allowRunCoalescing: wakeMustNotJoinActiveRun(input.activeExecutionRun.contextSnapshot, input.contextSnapshot, interactionReceipt) ? false : input.allowRunCoalescing, sameDurableActor, isSameExecutionAgent, @@ -893,6 +892,7 @@ export function createAdmitWakeBehindIssueExecution(deps: { companyId: input.companyId, agentId: input.agentId, issueId: input.issueId, + chatConversationId: readWakeChatConversationId(input.contextSnapshot), ...(input.durableReceipt ? { durableActor: { diff --git a/server/src/modules/wake-queue/domain/context.ts b/server/src/modules/wake-queue/domain/context.ts index 67cc586fdca..4da2346fa63 100644 --- a/server/src/modules/wake-queue/domain/context.ts +++ b/server/src/modules/wake-queue/domain/context.ts @@ -4,8 +4,8 @@ // shape; this file holds only the small pieces that are specific to // building the release half's promoted-run snapshot. -import { extractWakeCommentIds, WAKE_COMMENT_IDS_KEY } from "../../run-dispatch/index.js"; -import { readNonEmptyString } from "./values.js"; +import { extractWakeCommentIds, WAKE_COMMENT_IDS_KEY, wakeChatConversationsConflict } from "../../run-dispatch/index.js"; +import { parseObject, readNonEmptyString } from "./values.js"; const PAPERCLIP_WAKE_PAYLOAD_KEY = "paperclipWake"; const PAPERCLIP_WAKE_COMMENT_KEY = "paperclipWakeComment"; @@ -69,6 +69,22 @@ export function hasInteractionContinuationWakeContext(contextSnapshot: Record readNonEmptyString(contextSnapshot[key])); } +/** + * Whether a wake must stay out of the active issue run. Each resolved card has + * its own immutable decision and continuation, so it never merges with + * comments or another card, in either arrival order. A reviewed chat run + * attests exactly one conversation, so a reply from another conversation on + * the same issue queues as its own wake. + */ +export function wakeMustNotJoinActiveRun( + activeRunContext: unknown, + wakeContext: Record, + interactionReceipt: boolean, +): boolean { + return interactionReceipt || Boolean(parseObject(activeRunContext).interactionId) || + wakeChatConversationsConflict(activeRunContext, wakeContext); +} + function isInteractionResolutionWakePayload(payload: Record | null | undefined): boolean { return readNonEmptyString(payload?.mutation) === "interaction"; } diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index a25bcb27dfb..e873bb16cca 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -79,6 +79,7 @@ import { NativeChatReviewPresentationContentionError, } from "../native-runtime/native-chat-review-presentation.js"; import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js"; +import { CHAT_CONVERSATION_ID_KEY } from "../../modules/run-dispatch/index.js"; import { projectSafeChatPublication } from "../chat-publication-projection.js"; import { @@ -1598,6 +1599,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti } : {}), taskKey: source.taskKey, source: `chat:${source.provider}`, + [CHAT_CONVERSATION_ID_KEY]: source.conversationId, wakeCommentId: source.commentIds.at(-1)!, wakeCommentIds: source.commentIds, retryOfRunId: source.failedRunId, diff --git a/server/src/services/chat-channels/inbound/wakeups.ts b/server/src/services/chat-channels/inbound/wakeups.ts index 54269fa8aab..a839c9b09b9 100644 --- a/server/src/services/chat-channels/inbound/wakeups.ts +++ b/server/src/services/chat-channels/inbound/wakeups.ts @@ -362,6 +362,7 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) { requestedByActorId: request.requestedByActorId, taskKey: context.issue.identifier, wakeCommentId: request.commentId, + chatConversationId: context.conversation.id, attachmentOmissionReasons: claimed.payload.attachmentOmissionReasons as | Record | undefined, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index abb2f520ba0..4a1df79a5b1 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -147,7 +147,7 @@ import { WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION, } from "../execution-workspace-policy.js"; import { readContinuationAttempt } from "../recovery/index.js"; -import { allowsIssueInteractionWake } from "../../modules/run-dispatch/index.js"; +import { allowsIssueInteractionWake, wakeChatConversationsConflict } from "../../modules/run-dispatch/index.js"; import { withAgentStartLock } from "../agent-start-lock.js"; import { evaluateAgentInvokability, @@ -4414,10 +4414,12 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ? null : (sameScopeRunningRun ?? null))); - const coalescedTargetRun = filterZombieCoalesceTarget( - rawCoalescedTarget, - liveRunExecutions, - ); + // A reviewed chat run attests one conversation, so never fold another + // conversation's reply into it; that reply queues its own run instead. + const coalescedTargetRun = rawCoalescedTarget && + wakeChatConversationsConflict(rawCoalescedTarget.contextSnapshot, enrichedContextSnapshot) + ? null + : filterZombieCoalesceTarget(rawCoalescedTarget, liveRunExecutions); if (coalescedTargetRun) { const mergedContextSnapshot = mergeCoalescedContextSnapshot( diff --git a/server/src/services/issue-assignment-wakeup.ts b/server/src/services/issue-assignment-wakeup.ts index 61145438304..96ec5d96318 100644 --- a/server/src/services/issue-assignment-wakeup.ts +++ b/server/src/services/issue-assignment-wakeup.ts @@ -37,6 +37,9 @@ export function queueIssueAssignmentWakeup(input: { /** Closed, server-derived omission counts for provider attachments on the * exact wake comment. These are prompt diagnostics, never authorization. */ attachmentOmissionReasons?: Record | null; + /** External chat conversation the wake answers. The wake queue keeps each + * conversation in its own batch, because a reviewed chat run attests one. */ + chatConversationId?: string | null; rethrowOnError?: boolean; durableChatRequest?: DurableChatWakeupRequest; }) { @@ -63,6 +66,7 @@ export function queueIssueAssignmentWakeup(input: { source: input.contextSource, ...(input.taskKey ? { taskKey: input.taskKey } : {}), ...(input.wakeCommentId ? { wakeCommentId: input.wakeCommentId } : {}), + ...(input.chatConversationId ? { chatConversationId: input.chatConversationId } : {}), ...(input.wakeCommentId && input.attachmentOmissionReasons ? { externalAttachmentOmissions: [