From 8a4eaab2c082015568db3385db58ee94e94ce63c Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 11 Nov 2026 19:42:20 -0600 Subject: [PATCH] fix(heartbeat): deliver a person's answer that waited behind a turn that closed the task A person can answer an agent's question while the agent's turn is still running. The answer then waits on its own deferred wake. If that turn completed the task, release cancelled every deferred assignee wake on the done task, including the answer, even though the person's own queued messages reopen the task. The answer was dropped without any run receiving it. Treat an undelivered person's answer that was saved before the task was completed as owed input: reopen the done task straight to in_progress (an answer continues the assignee's own execution) and promote the answer wake. Answers saved after completion, tool-action outcomes, already-delivered answers, agent-requested wakes and cancelled tasks keep the old behavior. Eval steering.approval-delivery (oracle v7): also accept Acorn work that the worker's settling turn saved after the answer was saved, while still requiring exactly one later turn bound to the answer. Co-Authored-By: Claude Opus 5.5 --- ...artbeat-answer-behind-settling-run.test.ts | 128 ++++++++++++++++++ .../modules/wake-queue/adapters/postgres.ts | 14 +- .../modules/wake-queue/application/ports.ts | 8 +- .../wake-queue/application/use-cases.test.ts | 29 ++++ .../wake-queue/application/use-cases.ts | 14 ++ .../runner-e2e/chat-catalog-descriptors.json | 2 +- .../runtime-approval-queue.test.ts | 27 +++- .../runtime-approval-queue.ts | 25 +++- 8 files changed, 238 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..6f2c2f84d1d --- /dev/null +++ b/server/src/__tests__/heartbeat-answer-behind-settling-run.test.ts @@ -0,0 +1,128 @@ +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, + 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 } = {}) { + 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)); + gate.onTurn = async () => { 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 }; + } + + 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); +}); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 9fb73e8ee39..08cc5036555 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,15 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }; }, + async isOwedInteractionAnswer({ companyId, issueId, payload }) { + const response = await readQueuedInteractionResponse(tx as unknown as Db, companyId, issueId, payload); + if (!response || response.toolActionOutcome || !response.comment.authorUserId) return false; + 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); + return !!task && (!task.completedAt || answeredAt.getTime() <= task.completedAt.getTime()); + }, + async getCommentSelfAuthorship({ companyId, issueId, finishingRunId, commentIds }) { const rows = await tx .select({ createdByRunId: issueComments.createdByRunId }) @@ -500,8 +510,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..c9565683e5e 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -165,6 +165,11 @@ export interface WakeQueueTransaction { reason: string | null; releasePolicy: unknown; }>; + /** + * True when 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. + */ + isOwedInteractionAnswer(input: { companyId: string; issueId: string; payload: Record }): Promise; getCommentSelfAuthorship(input: { companyId: string; issueId: string; @@ -179,7 +184,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..ef5366b7364 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 })), + isOwedInteractionAnswer: vi.fn(async () => false), isCompletedDelegationMention: vi.fn(async () => false), reopenIssue: vi.fn(async () => null), claimDeferredWakeForPromotion: vi.fn(async () => true), @@ -594,6 +595,34 @@ describe("releaseIssueExecution", () => { expect(result.outcome.kind).toBe("promoted"); }); + it.each([ + ["owed answer on a done task", "done", "user", true, "promoted"], + ["answer already delivered or saved after completion", "done", "user", false, "released"], + ["answer on a cancelled task", "cancelled", "user", true, "released"], + ["agent-requested interaction wake", "done", "agent", true, "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), + isOwedInteractionAnswer: 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.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..63458192310 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -374,6 +374,19 @@ 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. + let reopenForAnswer = false; + if ( + !shouldReopen && + currentIssue.status === "done" && + workingCandidate.requestedByActorType === "user" && + workingCandidate.agentId === currentIssue.assigneeAgentId && + await ports.transaction.isOwedInteractionAnswer({ companyId: run.companyId, issueId: currentIssue.id, payload: workingCandidate.payload }) + ) 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 +425,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" });