diff --git a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts index a062a868858..855d99c4286 100644 --- a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts +++ b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts @@ -1107,10 +1107,13 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { }, 120_000); it.each([ - { caseName: "delivers explicit agent feedback after completion", targetAssignee: true, terminalStatus: "done", explicitResume: true }, - { caseName: "cancels an assignee continuation without resume intent on completed work", targetAssignee: true, terminalStatus: "done", explicitResume: false }, - { caseName: "cancels an assignee continuation on cancelled work", targetAssignee: true, terminalStatus: "cancelled", explicitResume: true }, - ] as const)("$caseName", async ({ targetAssignee, terminalStatus, explicitResume }) => { + { caseName: "delivers explicit agent feedback after completion", targetAssignee: true, terminalStatus: "done", explicitResume: true, commenterCreatedTask: false }, + // The delegating agent commented while the run it was waiting on was active. + // It receives the completed result; its comment does not reopen the task. + { caseName: "keeps a task done for its delegator's comment posted during the closing run", targetAssignee: true, terminalStatus: "done", explicitResume: true, commenterCreatedTask: true }, + { caseName: "cancels an assignee continuation without resume intent on completed work", targetAssignee: true, terminalStatus: "done", explicitResume: false, commenterCreatedTask: false }, + { caseName: "cancels an assignee continuation on cancelled work", targetAssignee: true, terminalStatus: "cancelled", explicitResume: true, commenterCreatedTask: false }, + ] as const)("$caseName", async ({ targetAssignee, terminalStatus, explicitResume, commenterCreatedTask }) => { const gateway = await createControlledGatewayServer(); const companyId = randomUUID(); const assigneeAgentId = randomUUID(); @@ -1119,7 +1122,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; const heartbeat = heartbeatService(db); const targetAgentId = targetAssignee ? assigneeAgentId : mentionedAgentId; - const shouldReopen = targetAssignee && terminalStatus === "done" && explicitResume; + const shouldReopen = targetAssignee && terminalStatus === "done" && explicitResume && !commenterCreatedTask; const commentingAgentId = targetAssignee ? mentionedAgentId : assigneeAgentId; const wakeReason = "issue_commented"; @@ -1170,7 +1173,10 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { }, ]); - await db.insert(issues).values(wakeIssueRow(issueId, companyId, "Agent feedback at completion boundary", assigneeAgentId, issuePrefix)); + await db.insert(issues).values({ + ...wakeIssueRow(issueId, companyId, "Agent feedback at completion boundary", assigneeAgentId, issuePrefix), + ...(commenterCreatedTask ? { createdByAgentId: commentingAgentId } : {}), + }); const firstRun = await heartbeat.wakeup(assigneeAgentId, { source: "assignment", diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 5d822efd755..59dce5b560f 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -472,6 +472,33 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru return { allSelfAuthored: rows.length > 0 && rows.every((row) => row.createdByRunId === finishingRunId) }; }, + async isWatcherCommentDuringFinishingRun({ companyId, issueId, finishingRunId, commentIds }) { + if (run.companyId !== companyId || run.id !== finishingRunId || !run.startedAt) return false; + const uniqueCommentIds = [...new Set(commentIds)]; + if (uniqueCommentIds.length === 0) return false; + const [task] = await tx.select({ createdByAgentId: issues.createdByAgentId, parentId: issues.parentId }) + .from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, issueId))); + if (!task) return false; + const watchers = new Set(task.createdByAgentId ? [task.createdByAgentId] : []); + if (task.parentId) { + const [parent] = await tx.select({ assigneeAgentId: issues.assigneeAgentId }).from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.id, task.parentId))); + if (parent?.assigneeAgentId) watchers.add(parent.assigneeAgentId); + } + watchers.delete(run.agentId); + if (watchers.size === 0) return false; + const comments = await tx.select({ authorType: issueComments.authorType, authorAgentId: issueComments.authorAgentId, + authorUserId: issueComments.authorUserId, createdAt: issueComments.createdAt }) + .from(issueComments).where(and(eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), + inArray(issueComments.id, uniqueCommentIds), isNull(issueComments.deletedAt))); + const startedAt = run.startedAt.getTime(); + return comments.length === uniqueCommentIds.length && comments.every((comment) => + // Older rows carry no author type; an agent author and no person is the same fact. + (comment.authorType === "agent" || comment.authorType === null) && + comment.authorUserId === null && comment.authorAgentId !== null && + watchers.has(comment.authorAgentId) && comment.createdAt.getTime() >= startedAt); + }, + async isCompletedDelegationMention({ companyId, issueId, finishingRunId, wakeAgentId, commentIds }) { if ( run.companyId !== companyId || run.id !== finishingRunId || diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index e55c2086b8c..57c53691f82 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -185,6 +185,18 @@ export interface WakeQueueTransaction { wakeAgentId: string; commentIds: string[]; }): Promise; + /** + * True when every comment was posted by the agent watching the task (its + * creator or its parent's assignee) while the finishing run was active. That + * agent receives the completed result and can follow up; its comment does + * not need to reopen the task the run just closed. + */ + isWatcherCommentDuringFinishingRun(input: { + companyId: string; + issueId: string; + finishingRunId: string; + commentIds: string[]; + }): Promise; /** Reopens to todo, or straight to in_progress for an owed answer, which continues the assignee's own execution. */ reopenIssue(input: { companyId: string; issueId: string; runId: string; status?: "todo" | "in_progress" }): Promise; /** Verifies a Done onboarding parent's completion wake against its own completed children. */ 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 9f44d6e151e..4e6bdf7566c 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -115,6 +115,7 @@ function createFakeTransaction(overrides: Partial = {}): W releasePolicy: null, })), getCommentSelfAuthorship: vi.fn(async () => ({ allSelfAuthored: false })), + isWatcherCommentDuringFinishingRun: vi.fn(async () => false), owedInteractionAnswer: vi.fn(async (): Promise<"deliver" | "blocked" | null> => null), isCompletedDelegationMention: vi.fn(async () => false), reopenIssue: vi.fn(async () => null), @@ -643,7 +644,7 @@ describe("releaseIssueExecution", () => { expect(transaction.cancelDeferredWake).toHaveBeenCalledWith(expect.objectContaining({ wakeId: "next" })); }); - it.each(["done_live", "cancelled_live", "done_missing", "done_self", "done_no_resume", "done_untracked_comment"])("handles explicit agent feedback after completion: %s", async (scenario) => { + it.each(["done_live", "cancelled_live", "done_missing", "done_self", "done_no_resume", "done_untracked_comment", "done_watcher_during_run"])("handles explicit agent feedback after completion: %s", async (scenario) => { const commentIds = ["accepted-agent-feedback"]; const queue = [wakeCandidate({ agentId: ISSUE.assigneeAgentId!, @@ -661,6 +662,7 @@ describe("releaseIssueExecution", () => { containedSelfAuthoredComment: scenario === "done_self", })), getCommentSelfAuthorship: vi.fn(async () => ({ allSelfAuthored: scenario === "done_self" })), + isWatcherCommentDuringFinishingRun: vi.fn(async () => scenario === "done_watcher_during_run"), reopenIssue: vi.fn(async () => ({ ...ISSUE, status: "todo" })), }); const release = createReleaseIssueExecution({ diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index ae47f878215..f8a939e13f5 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -363,15 +363,25 @@ async function promoteDeferredWake( finishingRunId: run.id, commentIds: workingCandidate.deferredCommentIds, }); + const agentFeedbackAfterCompletion = + currentIssue.status === "done" && + workingCandidate.agentId === currentIssue.assigneeAgentId && + workingCandidate.requestedByActorType === "agent" && + workingCandidate.deferredContextSeed.resumeIntent === true && + workingCandidate.queuedCommentIds.length > 0; shouldReopen = !selfAuthorship.allSelfAuthored && (workingCandidate.requestedByActorType === "user" || workingCandidate.wakeReason === "issue_reopened_via_comment" || - (currentIssue.status === "done" && - workingCandidate.agentId === currentIssue.assigneeAgentId && - workingCandidate.requestedByActorType === "agent" && - workingCandidate.deferredContextSeed.resumeIntent === true && - workingCandidate.queuedCommentIds.length > 0)); + // The agent watching the task posted it while the run that closed the + // task was active. That agent receives the completed result and can + // follow up from it; reopening would only repeat finished work. A + // person's message (handled above) always reopens. + (agentFeedbackAfterCompletion && run.agentId === currentIssue.assigneeAgentId && + !await ports.transaction.isWatcherCommentDuringFinishingRun({ + companyId: run.companyId, issueId: currentIssue.id, finishingRunId: run.id, + commentIds: workingCandidate.queuedCommentIds, + }))); } // A person's answer saved while the task was still open is owed input. The