From 2a1620f1434ee88764af1587b18576d3a1aebfa8 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 20:33:48 -0600 Subject: [PATCH] fix(heartbeat): deliver a person's answer that waited behind a turn that closed the task (#20996) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - An agent can ask the person a question on its task, and the person answers it in the task chat > - The person can answer while the agent's turn is still running. The answer then waits for that turn to finish > - If that turn completes the task, the release step cancels every waiting wake for the assignee on the done task. The person's own queued messages reopen the task, but the answer was dropped and no turn received it > - A person who answers an agent's question expects the agent to get the answer. Losing it without a trace is a silent failure > - This pull request treats an answer saved before the task was completed as owed input. It reopens the task and delivers the answer once > - The benefit is that a person's answer is never dropped because the agent finished its turn first ## Linked Issues or Issue Description **What happened?** The paid chat eval `steering.approval-delivery` showed it. The person answered the agent's naming question while the agent's turn was running. The turn completed the task. The answer's own wake was then cancelled with "Deferred execution wake no longer applies to a terminal task". No turn received the answer. **Steps to reproduce** 1. Start a turn for the assignee on a task. 2. While the turn runs, answer the agent's question on that task. 3. Let the turn mark the task done. 4. The answer's wake is cancelled, and no run receives it. **Expected behavior** The answer reaches the assignee exactly once. An answer saved after the task was completed is still dropped, as before. Refs #20981, #19331 ## What Changed - `wake-queue/application/use-cases.ts`: before cancelling a deferred assignee wake on a done task, check whether it carries a person's owed answer. If it does, reopen the task to `in_progress` and promote the wake. An answer continues the assignee's own execution, so the existing continuation status gate applies unchanged. - `wake-queue/adapters/postgres.ts`: add `owedInteractionAnswer`. It applies only to an undelivered response that a person resolved, that is not a tool action, and that was saved at or before the task's `completedAt`. It returns `blocked` when unresolved blockers forbid `in_progress`. `reopenIssue` takes an optional target status. - The blocked case cancels the answer wake with an explicit reason and does not reopen the task, so the release step does not fail. A regression test reproduces the earlier throw (fix for the Greptile P1). - `heartbeat-answer-behind-settling-run.test.ts`: an end-to-end heartbeat test. It is red on master (the answer wake is cancelled and no run is bound to it) and green here (one run is bound and succeeds). A control test shows that an answer saved after completion is still dropped. - `use-cases.test.ts`: unit cases for an owed answer, a not-owed answer, a cancelled task, and an agent-requested wake. - Eval `steering.approval-delivery` oracle v7: also accept the Acorn document when the worker's settling turn saved it after the answer was saved. The eval still requires exactly one later turn bound to the answer. Calibration tests cover the retained shape, reworded text, a pre-answer save, the wrong name, another agent, a stale revision, and a missing bound turn. ## Verification - `cd server && npx vitest run src/__tests__/heartbeat-answer-behind-settling-run.test.ts src/modules/wake-queue src/__tests__/heartbeat-blocked-question-answer.test.ts` (351 passed) - Red check: with the `server/src/modules` changes reverted, the new delivery test fails. - `cd server && npx vitest run src/__tests__/heartbeat-comment-wake-batching.test.ts src/__tests__/question-response-delivery.test.ts` passed. In `issue-queued-comments-routes.test.ts` all 138 tests passed, but a cleanup hook timed out locally. CI covers this file. - `npx vitest run --config tests/runner-e2e/vitest.config.ts tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.test.ts tests/runner-e2e/chat-catalog-consistency.test.ts` (58 passed) - `tsc --noEmit` passes for `server` and `tests/runner-e2e`. - Paid single-case run 43136466619 on 8a4eaab2c0 passed: one turn was bound to the answer, there were 5 runs within the bound, and the task was done. The later blocker commit adds only a blocker read on that path. ## Risks - Behavior change: a done task reopens when a person's answer was saved before completion and no run received it. In that case the agent gets one more turn. Answers saved after completion, cancelled tasks, tool-action outcomes, and agent-requested wakes do not change. - If the agent already read the answer by itself, the extra turn sees the finished work and completes again. That is one extra turn, and the answer is not repeated. - Known gap: the answer is still dropped when the task has unresolved blockers. This happens when a blocker reopens during the turn. Such a task cannot move to `in_progress`, and an answer turn cannot run on any other status. In this case the release step cancels the answer wake with the reason "Queued answer cannot continue a task with unresolved blockers", keeps the task done, and continues to the next wake. This is not a regression, because master dropped the answer in every case. But it does not meet the goal that a person's answer is never lost. It is a separate follow-up. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- ...artbeat-answer-behind-settling-run.test.ts | 149 ++++++++++++++++++ .../modules/wake-queue/adapters/postgres.ts | 17 +- .../modules/wake-queue/application/ports.ts | 9 +- .../wake-queue/application/use-cases.test.ts | 49 ++++++ .../wake-queue/application/use-cases.ts | 26 +++ .../runner-e2e/chat-catalog-descriptors.json | 2 +- .../runtime-approval-queue.test.ts | 27 +++- .../runtime-approval-queue.ts | 25 ++- 8 files changed, 295 insertions(+), 9 deletions(-) create mode 100644 server/src/__tests__/heartbeat-answer-behind-settling-run.test.ts diff --git a/server/src/__tests__/heartbeat-answer-behind-settling-run.test.ts b/server/src/__tests__/heartbeat-answer-behind-settling-run.test.ts new file mode 100644 index 00000000000..964d6a54639 --- /dev/null +++ b/server/src/__tests__/heartbeat-answer-behind-settling-run.test.ts @@ -0,0 +1,149 @@ +import { randomUUID } from "node:crypto"; +import { and, asc, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + agentWakeupRequests, + authUsers, + companies, + companyMemberships, + createDb, + heartbeatRuns, + issueComments, + issueRelations, + issueThreadInteractions, + issues, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.ts"; + +const gate = vi.hoisted(() => ({ release: () => {}, onTurn: async () => {}, turns: 0 })); +const mockAdapterExecute = vi.hoisted(() => vi.fn(async () => { + gate.turns += 1; + if (gate.turns === 1) await new Promise(resolve => { gate.release = resolve; }); + // Each turn finishes the task, as the worker in the eval did. + await gate.onTurn(); + return { exitCode: 0, signal: null, timedOut: false, errorMessage: null, summary: "Turn finished.", provider: "test", model: "test-model" }; +})); + +vi.mock("../adapters/index.ts", async () => { + const actual = await vi.importActual("../adapters/index.ts"); + return { ...actual, getServerAdapter: vi.fn(() => ({ supportsLocalAgentJwt: false, execute: mockAdapterExecute })) }; +}); + +const support = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; + +async function waitFor(check: () => Promise, timeoutMs = 30_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { if (await check()) return; await new Promise(r => setTimeout(r, 100)); } + throw new Error("Timed out waiting for condition"); +} + +describeEmbeddedPostgres("an answer saved while the worker settles its turn", () => { + let db!: ReturnType; + let heartbeat!: ReturnType; + let tempDb: Awaited> | null = null; + const userId = "answer-behind-settling-user"; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-answer-behind-settling-"); + db = createDb(tempDb.connectionString); + await db.insert(authUsers).values({ id: userId, name: "Board", email: "board@example.test", createdAt: new Date(), updatedAt: new Date() }); + heartbeat = heartbeatService(db); + }, 30_000); + + afterAll(async () => { + await heartbeat?.drainActiveRunExecutions(); + await db?.$client.end(); + await tempDb?.cleanup(); + }); + + /** + * Morgan's turn is running while the person answers its question and the chat lead relays the answer as a + * task comment; both wait behind the turn. The turn then completes the task without reading the answer. + */ + async function settleWithQueuedAnswer(options: { answerAfterCompletion?: boolean; blockerReopens?: boolean } = {}) { + const companyId = randomUUID(), agentId = randomUUID(), leadId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Answer fixture", issuePrefix: `A${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: userId }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" }); + for (const [id, name] of [[agentId, "Morgan"], [leadId, "CEO"]] as const) { + await db.insert(agents).values({ id, companyId, name, role: "engineer", status: "active", adapterType: "codex_local", adapterConfig: {}, permissions: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }); + } + await db.insert(issues).values({ id: issueId, companyId, title: "Website naming task", status: "todo", assigneeAgentId: agentId, responsibleUserId: userId }); + const [question] = await db.insert(issueThreadInteractions).values({ companyId, issueId, createdByAgentId: agentId, kind: "ask_user_questions", status: "pending", + continuationPolicy: "wake_assignee", title: "Website product name", + payload: { version: 1, questions: [{ id: "name", prompt: "Which product name?", selectionMode: "single", options: [{ id: "acorn", label: "Acorn" }, { id: "birch", label: "Birch" }] }] } }).returning(); + + gate.turns = 0; + const answeredAt = new Date(), completedAt = new Date(answeredAt.getTime() + (options.answerAfterCompletion ? -60_000 : 60_000)); + const blockerId = randomUUID(); + gate.onTurn = async () => { + // A blocker can reopen while the turn runs; the turn still marks this task done. + if (options.blockerReopens && gate.turns === 1) { + await db.insert(issues).values({ id: blockerId, companyId, title: "Brand review", status: "todo", responsibleUserId: userId }); + await db.insert(issueRelations).values({ companyId, issueId: blockerId, relatedIssueId: issueId, type: "blocks" }); + } + await db.update(issues).set({ status: "done", completedAt }).where(eq(issues.id, issueId)); + }; + const first = await heartbeat.wakeup(agentId, { source: "assignment", triggerDetail: "system", reason: "issue_assigned", + payload: { issueId }, contextSnapshot: { issueId, taskId: issueId, wakeReason: "issue_assigned" }, requestedByActorType: "system", requestedByActorId: null }); + expect(first).not.toBeNull(); + await waitFor(async () => gate.turns === 1); + + // The person answers while the worker is still settling its turn. + await db.update(issueThreadInteractions).set({ status: "answered", resolvedByUserId: userId, resolvedAt: answeredAt, + result: { version: 1, answers: [{ questionId: "name", optionIds: ["acorn"] }] } }).where(eq(issueThreadInteractions.id, question!.id)); + const answerContext = { issueId, taskId: issueId, mutation: "interaction", interactionId: question!.id, interactionKind: "ask_user_questions", + interactionStatus: "answered", wakeReason: "issue_commented", source: "issue.interaction.respond" }; + await heartbeat.wakeup(agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", payload: answerContext, + contextSnapshot: answerContext, requestedByActorType: "user", requestedByActorId: userId }); + // The chat lead relays the answer as its own comment on the task. + const [relay] = await db.insert(issueComments).values({ companyId, issueId, authorAgentId: leadId, body: "The product name is Acorn." }).returning(); + await heartbeat.wakeup(agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId, commentId: relay!.id }, contextSnapshot: { issueId, taskId: issueId, commentId: relay!.id, wakeCommentId: relay!.id, wakeReason: "issue_commented" }, + requestedByActorType: "agent", requestedByActorId: leadId }); + + gate.release(); + const settled = async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId))).every(r => ["succeeded", "failed", "cancelled", "timed_out"].includes(r.status)); + await waitFor(async () => (await settled()) && (await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution")))).length === 0, 60_000); + await heartbeat.drainActiveRunExecutions(); + await waitFor(settled, 60_000); + return { question: question!, issueId, agentId, blockerId }; + } + + it("delivers a person's answer saved before the worker closed the task, without replaying it", async () => { + const { question, agentId } = await settleWithQueuedAnswer(); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)).orderBy(asc(heartbeatRuns.createdAt)); + const bound = runs.filter(r => (r.contextSnapshot as Record | null)?.interactionId === question.id); + expect(bound.map(r => ({ status: r.status, started: !!r.startedAt }))).toEqual([{ status: "succeeded", started: true }]); + const [answerWake] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.agentId, agentId), + sql`${agentWakeupRequests.payload}->>'interactionId' = ${question.id}`)); + expect(answerWake?.status).not.toBe("cancelled"); + expect(answerWake?.runId).toBe(bound[0]!.id); + }, 120_000); + + it("still drops an answer saved only after the task was completed", async () => { + const { question, agentId } = await settleWithQueuedAnswer({ answerAfterCompletion: true }); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs.filter(r => (r.contextSnapshot as Record | null)?.interactionId === question.id)).toEqual([]); + const [answerWake] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.agentId, agentId), + sql`${agentWakeupRequests.payload}->>'interactionId' = ${question.id}`)); + expect(answerWake).toMatchObject({ status: "cancelled", error: "Deferred execution wake no longer applies to a terminal task" }); + }, 120_000); + + it("settles without reopening when a blocker reopened during the turn forbids in_progress", async () => { + const { question, agentId, issueId, blockerId } = await settleWithQueuedAnswer({ blockerReopens: true }); + const [task] = await db.select({ status: issues.status }).from(issues).where(eq(issues.id, issueId)); + expect(task?.status).toBe("done"); + const [answerWake] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.agentId, agentId), + sql`${agentWakeupRequests.payload}->>'interactionId' = ${question.id}`)); + expect(answerWake).toMatchObject({ status: "cancelled", error: "Queued answer cannot continue a task with unresolved blockers", runId: null }); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs.map(r => r.status)).toEqual(["succeeded"]); + expect(blockerId).toBeTruthy(); + }, 120_000); +}); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 9fb73e8ee39..6b507505e9b 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -34,6 +34,7 @@ import { nativeRunFinalizations, } from "@paperclipai/db"; import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; +import { readQueuedInteractionResponse } from "../../../services/queued-interaction-response.js"; import { legacyExecutionNeedsReconciliationWithEvidence } from "../../../services/legacy-execution-recovery.js"; import { authorizeFailedChatRunRetryWake, @@ -450,6 +451,18 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }; }, + async owedInteractionAnswer({ companyId, issueId, payload }) { + const response = await readQueuedInteractionResponse(tx as unknown as Db, companyId, issueId, payload); + if (!response || response.toolActionOutcome || !response.comment.authorUserId) return null; + const [task] = await tx.select({ completedAt: issues.completedAt }).from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.id, issueId))); + const answeredAt = new Date(response.comment.createdAt); + if (!task || (task.completedAt && answeredAt.getTime() > task.completedAt.getTime())) return null; + // in_progress is refused while blockers are unresolved, and reopening would throw inside this drain. + const readiness = await issuesSvc.getDependencyReadiness(issueId, tx); + return readiness.unresolvedBlockerIssueIds.length > 0 ? "blocked" : "deliver"; + }, + async getCommentSelfAuthorship({ companyId, issueId, finishingRunId, commentIds }) { const rows = await tx .select({ createdByRunId: issueComments.createdByRunId }) @@ -500,8 +513,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru return isCompletedOnboardingHandoffWake(tx, input); }, - async reopenIssue({ companyId, issueId }) { - const updated = await issuesSvc.updateForCompany(issueId, companyId, { status: "todo", executionState: null }, tx); + async reopenIssue({ companyId, issueId, status = "todo" }) { + const updated = await issuesSvc.updateForCompany(issueId, companyId, { status, executionState: null }, tx); return updated ? toIssueSnapshot(updated as unknown as IssueRow) : null; }, diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index 7cf3cc4e7c8..e55c2086b8c 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -165,6 +165,12 @@ export interface WakeQueueTransaction { reason: string | null; releasePolicy: unknown; }>; + /** + * Whether the wake carries a person's saved answer to an interaction on this task that no run has received yet and + * that was saved before the task was completed. Such an answer is owed input, like the person's own queued message. + * "deliver" when the task can run now; "blocked" when unresolved blockers keep it from in_progress; null otherwise. + */ + owedInteractionAnswer(input: { companyId: string; issueId: string; payload: Record }): Promise<"deliver" | "blocked" | null>; getCommentSelfAuthorship(input: { companyId: string; issueId: string; @@ -179,7 +185,8 @@ export interface WakeQueueTransaction { wakeAgentId: string; commentIds: string[]; }): Promise; - reopenIssue(input: { companyId: string; issueId: string; runId: 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. */ isCompletedOnboardingHandoffWake(input: { companyId: string; issueId: string; agentId: string; reason: string | null; contextSnapshot: Record }): 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 4e2dcd37496..9f44d6e151e 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 })), + owedInteractionAnswer: vi.fn(async (): Promise<"deliver" | "blocked" | null> => null), isCompletedDelegationMention: vi.fn(async () => false), reopenIssue: vi.fn(async () => null), claimDeferredWakeForPromotion: vi.fn(async () => true), @@ -594,6 +595,54 @@ describe("releaseIssueExecution", () => { expect(result.outcome.kind).toBe("promoted"); }); + it.each([ + ["owed answer on a done task", "done", "user", "deliver", "promoted"], + ["answer already delivered or saved after completion", "done", "user", null, "released"], + ["answer on a cancelled task", "cancelled", "user", "deliver", "released"], + ["agent-requested interaction wake", "done", "agent", "deliver", "released"], + ] as const)("handles a person's answer queued behind a turn that closed the task: %s", async (_label, status, actor, owed, outcome) => { + const queue = [wakeCandidate({ agentId: ISSUE.assigneeAgentId!, requestedByActorType: actor, + payload: { issueId: ISSUE.id, mutation: "interaction", interactionId: "11111111-1111-4111-8111-111111111111", interactionStatus: "answered" } })]; + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async () => queue.shift() ?? null), + owedInteractionAnswer: vi.fn(async () => owed), + reopenIssue: vi.fn(async () => ({ ...ISSUE, status: "in_progress" })), + }); + const release = createReleaseIssueExecution({ + issueLock: createFakeIssueLock(createFakeHost(), transaction, { ...ISSUE, status }), + recovery: createFakeRecovery(), + }); + const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() }); + expect(result.outcome.kind).toBe(outcome); + if (outcome === "promoted") { + expect(transaction.reopenIssue).toHaveBeenCalledWith(expect.objectContaining({ status: "in_progress" })); + expect(transaction.cancelDeferredWake).not.toHaveBeenCalled(); + } else { + expect(transaction.reopenIssue).not.toHaveBeenCalled(); + expect(transaction.cancelDeferredWake).toHaveBeenCalledWith(expect.objectContaining({ reason: "Deferred execution wake no longer applies to a terminal task" })); + } + }); + + it("cancels an owed answer without reopening or failing the drain while blockers forbid in_progress", async () => { + const queue = [wakeCandidate({ id: "answer", agentId: ISSUE.assigneeAgentId!, requestedByActorType: "user", + payload: { issueId: ISSUE.id, mutation: "interaction", interactionId: "11111111-1111-4111-8111-111111111111", interactionStatus: "answered" } }), + wakeCandidate({ id: "next", agentId: ISSUE.assigneeAgentId!, requestedByActorType: "agent" })]; + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async () => queue.shift() ?? null), + owedInteractionAnswer: vi.fn(async () => "blocked" as const), + }); + const release = createReleaseIssueExecution({ + issueLock: createFakeIssueLock(createFakeHost(), transaction, { ...ISSUE, status: "done" }), + recovery: createFakeRecovery(), + }); + const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() }); + expect(result.outcome.kind).toBe("released"); + expect(transaction.reopenIssue).not.toHaveBeenCalled(); + expect(transaction.cancelDeferredWake).toHaveBeenCalledWith(expect.objectContaining({ wakeId: "answer", reason: "Queued answer cannot continue a task with unresolved blockers" })); + // The drain continues past the answer to the next queued wake. + 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) => { const commentIds = ["accepted-agent-feedback"]; const queue = [wakeCandidate({ diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index 3feacde9cb7..ae47f878215 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -374,6 +374,31 @@ async function promoteDeferredWake( workingCandidate.queuedCommentIds.length > 0)); } + // A person's answer saved while the task was still open is owed input. The + // worker may have finished its turn without reading it; closing the task + // must not silently drop it. It reopens the task straight to in_progress, + // because an answer continues the assignee's own execution. While + // unresolved blockers forbid in_progress, an answer continuation cannot run + // (it requires in_progress), so the task stays done and the wake is + // cancelled with that reason; it never fails or stalls this drain. + let reopenForAnswer = false; + if ( + !shouldReopen && + currentIssue.status === "done" && + workingCandidate.requestedByActorType === "user" && + workingCandidate.agentId === currentIssue.assigneeAgentId + ) { + const owed = await ports.transaction.owedInteractionAnswer({ companyId: run.companyId, issueId: currentIssue.id, payload: workingCandidate.payload }); + if (owed === "blocked") { + await ports.transaction.cancelDeferredWake({ + companyId: run.companyId, wakeId: workingCandidate.id, agentId: workingCandidate.agentId, + reason: "Queued answer cannot continue a task with unresolved blockers", now: input.now, + }); + return null; + } + if (owed === "deliver") shouldReopen = reopenForAnswer = true; + } + // Agent continuations can outlive the work they addressed. Live, // non-self feedback with an explicit resume intent must still be read // after completion; it cannot revive a cancelled task. Other stale @@ -412,6 +437,7 @@ async function promoteDeferredWake( companyId: run.companyId, issueId: currentIssue.id, runId: run.id, + ...(reopenForAnswer ? { status: "in_progress" as const } : {}), }); if (reopened) { postCommitEffects.push({ diff --git a/tests/runner-e2e/chat-catalog-descriptors.json b/tests/runner-e2e/chat-catalog-descriptors.json index 7e25421f14b..4f1f3b95753 100644 --- a/tests/runner-e2e/chat-catalog-descriptors.json +++ b/tests/runner-e2e/chat-catalog-descriptors.json @@ -5451,7 +5451,7 @@ "chat-catalog-domains/runtime-support.ts", "chat-scenarios/steering.yaml" ], - "oracleVersion": "paperclip.chat-catalog.runtime.approval-delivery.v6", + "oracleVersion": "paperclip.chat-catalog.runtime.approval-delivery.v7", "coverageGaps": [], "expectedRunCount": 5, "minimumRunCount": 4, diff --git a/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.test.ts b/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.test.ts index c5b42ff9483..3b3d79baee6 100644 --- a/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.test.ts +++ b/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.test.ts @@ -1,5 +1,5 @@ -import { expect, it } from "vitest"; -import { approvalResponseListing, classifyApprovalSuccessors, gradeApprovalQueue, leadRelayEntries, ordinaryApprovalEntries, pendingInputYield } from "./runtime-approval-queue.js"; +import { describe, expect, it } from "vitest"; +import { acornWorkSource, approvalResponseListing, classifyApprovalSuccessors, gradeApprovalQueue, leadRelayEntries, ordinaryApprovalEntries, pendingInputYield } from "./runtime-approval-queue.js"; function valid() { const payload = { questionSet: { questions: [{ id: "name", prompt: "What is the product name?" }] } }; return { scenarioId: "steering.approval-delivery", companyId: "company", leadId: "ceo", agents: [], links: [], humanQueries: [], reply: "", tasks: [{ id: "website" }], activity: [], @@ -100,3 +100,26 @@ it("keeps only the person's own messages as editable ordinary entries", () => { expect(ordinaryApprovalEntries(queue, "human").map(row => row.comment.id)).toEqual(["mine"]); expect(leadRelayEntries(queue, "ceo").map(row => row.comment.id)).toEqual(["relay"]); }); + +// Retained runner-opencode attempt (gha-43129293316): the settling turn read the answered question itself and saved the +// Acorn document after the answer was saved; the answer must still reach one bound turn (product fix in this change). +describe("where the Acorn document came from", () => { + const answeredAt = "2026-11-12T01:03:19.592Z"; + const doc = (runId: string, createdAt: string, body = "# Acorn website delivery\n\nProduct name: **Acorn**", agent = "morgan") => ({ + revisions: [{ id: "rev", documentId: "doc", createdByRunId: runId, createdByAgentId: agent, createdAt, body }], documents: [{ id: "doc", latestRevisionId: "rev" }] }); + const source = (input: ReturnType, extra: Partial[0]> = {}) => acornWorkSource({ ...input, workerId: "morgan", + successorRunIds: ["bound", "relay"], settlingRunId: "settling", answeredAt, ...extra }); + it("accepts the bound turn's document", () => expect(source(doc("bound", "2026-11-12T01:05:30.000Z"))).toBe("delivered_turn")); + it("accepts the lead relay turn's document", () => expect(source(doc("relay", "2026-11-12T01:05:00.000Z"))).toBe("delivered_turn")); + it("accepts the settling turn's document saved after the answer (the retained shape)", () => expect(source(doc("settling", "2026-11-12T01:04:21.660Z"))).toBe("settling_turn_after_answer")); + it("accepts a reworded document that still names Acorn", () => expect(source(doc("settling", "2026-11-12T01:04:21.660Z", "Website outline for Acorn: hero, features, contact."))).toBe("settling_turn_after_answer")); + it("rejects a settling-turn document saved before the answer existed", () => expect(source(doc("settling", "2026-11-12T01:02:00.000Z"))).toBeNull()); + it("rejects a document that does not use the answered name", () => expect(source(doc("bound", "2026-11-12T01:05:30.000Z", "# Website delivery\n\nProduct name: Birch"))).toBeNull()); + it("rejects a name that only appears inside another word", () => expect(source(doc("bound", "2026-11-12T01:05:30.000Z", "Acorns are a placeholder."))).toBeNull()); + it("rejects a document saved by another agent", () => expect(source(doc("bound", "2026-11-12T01:05:30.000Z", undefined, "ceo"))).toBeNull()); + it("rejects a revision that is no longer the document's current one", () => { const d = doc("bound", "2026-11-12T01:05:30.000Z"); d.documents[0]!.latestRevisionId = "newer"; + expect(source(d)).toBeNull(); }); + it("rejects a document from an unrelated turn", () => expect(source(doc("other", "2026-11-12T01:05:30.000Z"))).toBeNull()); + it("rejects the settling turn when the answer time is unknown", () => expect(source(doc("settling", "2026-11-12T01:04:21.660Z"), { answeredAt: null })).toBeNull()); + it("still fails the case when the settling turn used the answer but no turn was bound to it", () => { const o = valid(); o.state.consumingRunCount = 0; expect(gradeApprovalQueue(o).passed).toBe(false); }); +}); diff --git a/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.ts b/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.ts index 238385b735a..f6b5aadc15e 100644 --- a/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.ts +++ b/tests/runner-e2e/chat-catalog-domains/runtime-approval-queue.ts @@ -52,6 +52,23 @@ export function classifyApprovalSuccessors(runs: SuccessorRun[], questionId: str } return { bound, relay, other }; } +type SavedRevision = { id?: unknown; documentId?: unknown; createdByRunId?: unknown; createdByAgentId?: unknown; createdAt?: unknown; body?: unknown }; +/** + * Where the current Acorn document came from. A later worker turn (the one bound to the answer, or the lead's relay + * turn) counts. So does the worker's settling turn, but only for a revision it saved after the answer was saved: an + * agent may read the answered question on its own task, and the answer still reaches exactly one bound turn later. + * A revision saved before the answer existed, by another agent, or no longer current does not count. + */ +export function acornWorkSource(input: { revisions: SavedRevision[]; documents: Array<{ id?: unknown; latestRevisionId?: unknown }>; workerId: string; + successorRunIds: string[]; settlingRunId: string; answeredAt: unknown }) { + const answered = typeof input.answeredAt === "string" ? Date.parse(input.answeredAt) : NaN; + const current = input.revisions.filter(row => row.createdByAgentId === input.workerId && typeof row.body === "string" && /\bAcorn\b/.test(row.body) + && input.documents.some(doc => doc.id === row.documentId && doc.latestRevisionId === row.id)); + if (current.some(row => typeof row.createdByRunId === "string" && input.successorRunIds.includes(row.createdByRunId))) return "delivered_turn" as const; + if (Number.isFinite(answered) && current.some(row => row.createdByRunId === input.settlingRunId && typeof row.createdAt === "string" && Date.parse(row.createdAt) > answered)) + return "settling_turn_after_answer" as const; + return null; +} export function gradeApprovalQueue(value: unknown) { const c = checks(), o = runtimeObservation(value); c.add("real-question-response-and-ordinary-queue-evidence", !!o, "The actual native question, human response, separate queue mutations and once-only consuming native input are required."); if (!o) return c.result(); @@ -118,7 +135,7 @@ export const runtimeApprovalQueueCases = [runtimeDefinition("steering.approval-d humanId: session.user.id, ordinaryQueueOwned: true, ordinaryQueuedBeforeAnswer: true }; }, gradeApprovalQueue, { setupPolicy: runtimeQueueCases[0]!.setupPolicy, verifySetup: runtimeQueueCases[0]!.verifySetup, - authentication: "human-session", expectedRunCount: 5, minimumRunCount: 4, attemptTimeoutMs: 900_000, oracleVersion: "paperclip.chat-catalog.runtime.approval-delivery.v6", + authentication: "human-session", expectedRunCount: 5, minimumRunCount: 4, attemptTimeoutMs: 900_000, oracleVersion: "paperclip.chat-catalog.runtime.approval-delivery.v7", sourceFiles: ["chat-catalog-domains/runtime-approval-queue.ts", "chat-catalog-domains/runtime-queue.ts", "chat-catalog-domains/runtime-busy-worker.ts", "chat-catalog-domains/runtime-support.ts", "chat-scenarios/steering.yaml"], async act(ctx, state) { const canonicalPhase = await pollUntil({ label: "Canonical naming answer guidance while real worker remains active", deadlineAt: Date.now() + 150_000, @@ -236,8 +253,10 @@ export const runtimeApprovalQueueCases = [runtimeDefinition("steering.approval-d }, async observe(ctx, state) { const o = await observeRuntime(ctx, state); const question = state.questionBefore as IssueThreadInteraction; state.questionAfter = (await ctx.input.api.get(`/api/issues/${(state.task as Entity).id}/interactions`)).find(row => row.id === question.id); - const successorIds = [state.consumerRunId, ...((state.relayRunIds as string[] | undefined) ?? [])]; - state.actualAcornResult = o.revisions.some(row => successorIds.includes(row.createdByRunId) && row.createdByAgentId === (state.worker as Entity).id && /Acorn/i.test(row.body ?? "") && o.documents.some(doc => doc.id === row.documentId && doc.latestRevisionId === row.id)); + const successorRunIds = [state.consumerRunId, ...((state.relayRunIds as string[] | undefined) ?? [])].filter((id): id is string => typeof id === "string"); + state.acornWorkSource = acornWorkSource({ revisions: o.revisions, documents: o.documents, workerId: (state.worker as Entity).id, successorRunIds, + settlingRunId: String(state.workerRunId), answeredAt: (state.questionAfter as IssueThreadInteraction | undefined)?.resolvedAt }); + state.actualAcornResult = state.acornWorkSource !== null; state.initialWorkRetained = o.revisions.some(row => row.createdByRunId === state.workerRunId && (row.body ?? "").includes(String(state.initialMarker))); return o; }, async verifyResult(ctx, state) { const source = ctx.input.page.url(), task = state.task as Entity; await ctx.input.page.goto(`/${ctx.input.fixtures.company.issuePrefix}/issues/${task.identifier ?? task.id}`, { waitUntil: "domcontentloaded" }); await ctx.input.page.reload({ waitUntil: "domcontentloaded" });