import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi, } from "vitest"; import { activityLog, agentWakeupRequests, agents, companies, createDb, goals, heartbeatRuns, issueComments, issueQuestionResponseDeliveries, issueThreadInteractions, issues, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { buildQuestionResponseDeliveryEnvelope, formatQuestionResponseSteeringMessage, questionResponseDeliveryService, } from "../services/question-response-delivery.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; const DIRECT_ADAPTER_TYPES = [ "acpx_local", "claude_local", "codex_local", "cursor_cloud", "cursor", "gemini_local", "grok_local", "hermes_gateway", "hermes_local", "kimi_local", "openclaw_gateway", "opencode_local", "pi_local", "process", "http", "external_test_adapter", ] as const; describeEmbeddedPostgres("question response delivery", () => { let db!: ReturnType; let tempDb: Awaited< ReturnType > | null = null; beforeAll(async () => { tempDb = await startEmbeddedPostgresTestDatabase( "paperclip-question-delivery-", ); db = createDb(tempDb.connectionString); }, 20_000); afterEach(async () => { await db.delete(issueQuestionResponseDeliveries); await db.delete(issueThreadInteractions); await db.delete(issueComments); await db.delete(activityLog); await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); await db.delete(issues); await db.delete(goals); await db.delete(agents); await db.delete(companies); }); afterAll(async () => { await tempDb?.cleanup(); }); async function seed( args: { adapterType?: string; runtimeMode?: "legacy" | "native"; sourceStatus?: string; successorStatus?: "queued" | "running"; sourceCommentBody?: string; attachSourceCommentToInteraction?: boolean; } = {}, ) { const companyId = randomUUID(); const agentId = randomUUID(); const goalId = randomUUID(); const issueId = randomUUID(); const sourceRunId = randomUUID(); const successorRunId = args.successorStatus ? randomUUID() : null; await db.insert(companies).values({ id: companyId, name: "Question delivery", issuePrefix: `Q${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, requireBoardApprovalForNewAgents: false, }); await db.insert(agents).values({ id: agentId, companyId, name: "Runner", role: "engineer", status: "active", adapterType: args.adapterType ?? "codex_local", adapterConfig: {}, runtimeConfig: {}, permissions: {}, }); await db.insert(goals).values({ id: goalId, companyId, title: "Test", level: "task", status: "active", }); await db.insert(issues).values({ id: issueId, companyId, goalId, title: "Deliver answers", status: "in_progress", priority: "medium", assigneeAgentId: agentId, }); await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId, invocationSource: "manual", status: args.sourceStatus ?? "succeeded", runtimeMode: args.runtimeMode ?? "native", driverKind: "codex", contextSnapshot: { issueId }, ...(args.sourceStatus === "running" ? { startedAt: new Date() } : { finishedAt: new Date() }), }); if (successorRunId && args.successorStatus) { await db.insert(heartbeatRuns).values({ id: successorRunId, companyId, agentId, invocationSource: "manual", status: args.successorStatus, runtimeMode: args.runtimeMode ?? "native", driverKind: "codex", contextSnapshot: { issueId }, ...(args.successorStatus === "running" ? { startedAt: new Date() } : {}), }); } const sourceCommentId = args.sourceCommentBody ? randomUUID() : null; if (sourceCommentId) { await db.insert(issueComments).values({ id: sourceCommentId, companyId, issueId, authorUserId: "external-user", body: args.sourceCommentBody!, }); } const interactionSvc = issueThreadInteractionService(db); const interaction = await interactionSvc.create( { id: issueId, companyId }, { kind: "ask_user_questions", continuationPolicy: "wake_assignee", sourceCommentId: args.attachSourceCommentToInteraction === false ? null : sourceCommentId, sourceRunId, payload: { version: 1, title: "Server choices", questions: [ { id: "purpose", prompt: "What is it for?", selectionMode: "single", required: true, options: [ { id: "custom", label: "Write an answer", freeText: true }, ], }, { id: "runtime", prompt: "Which runtime?", selectionMode: "single", required: true, options: [ { id: "node", label: "Node.js" }, { id: "bun", label: "Bun" }, ], }, { id: "features", prompt: "Which features?", selectionMode: "multi", allowOther: true, options: [ { id: "health", label: "Health check" }, { id: "logs", label: "Request logs" }, ], }, ], questionSet: { schema: "paperclip.question_set.v1", title: "Server choices", questions: [ { id: "purpose", header: "Purpose", prompt: "What is it for?", required: true, answerMode: "text", }, { id: "runtime", header: "Runtime", prompt: "Which runtime?", required: true, answerMode: "single_select", options: [ { id: "node", label: "Node.js" }, { id: "bun", label: "Bun" }, ], }, { id: "features", header: "Features", prompt: "Which features?", required: false, answerMode: "multi_select", options: [ { id: "health", label: "Health check" }, { id: "logs", label: "Request logs" }, ], customAnswer: { enabled: true, label: "Other" }, }, ], }, }, }, { agentId, runId: sourceRunId }, ); const answered = await interactionSvc.answerQuestions( { id: issueId, companyId, status: "in_progress" }, interaction.id, { answers: [ { questionId: "purpose", optionIds: [], otherText: "Internal API" }, { questionId: "runtime", optionIds: ["node"] }, { questionId: "features", optionIds: ["health", "logs"], otherText: "Metrics", }, ], }, { userId: "board-user" }, ); return { companyId, agentId, issueId, sourceRunId, sourceCommentId, successorRunId, interaction: answered, }; } it("queues an answer to a running successor instead of automatically steering", async () => { const seeded = await seed({ successorStatus: "running" }); const steer = vi.fn().mockResolvedValue({ turnId: "unexpected" }); const wakeup = vi.fn().mockImplementation(async () => { await db.insert(agentWakeupRequests).values({ companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", reason: "issue_commented", status: "deferred_issue_execution", idempotencyKey: `question-response:${seeded.interaction.id}`, payload: { issueId: seeded.issueId, mutation: "interaction", interactionId: seeded.interaction.id, interactionStatus: "answered" } }); return null; }); const service = questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer }); const first = await service.deliver(seeded.interaction.id); const second = await service.deliver(seeded.interaction.id); expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); expect(wakeup).toHaveBeenCalledWith(seeded.agentId, expect.objectContaining({ payload: expect.objectContaining({ interactionId: seeded.interaction.id, interactionStatus: "answered" }), })); expect(first).toMatchObject({ status: "fallback_queued", mode: "wake_fallback" }); expect(second).toMatchObject({ mode: "wake_fallback", duplicate: true }); }); it.each(["native", "legacy"] as const)( "never mixes an external-chat question response into another live chat run in %s mode", async (runtimeMode) => { const seeded = await seed({ runtimeMode, sourceStatus: "running", successorStatus: "running", sourceCommentBody: "External request whose complete trailing instructions must survive the response continuation.", attachSourceCommentToInteraction: false, }); if (!seeded.sourceCommentId) { throw new Error("Expected a source comment for the external-chat run"); } expect(seeded.interaction.sourceCommentId).toBeNull(); await db .update(heartbeatRuns) .set({ contextSnapshot: { issueId: seeded.issueId, source: "chat:slack", commentId: seeded.sourceCommentId, wakeCommentId: seeded.sourceCommentId, wakeCommentIds: [seeded.sourceCommentId], }, }) .where(eq(heartbeatRuns.id, seeded.sourceRunId)); await db .update(heartbeatRuns) .set({ contextSnapshot: { issueId: seeded.issueId, source: "chat:telegram", wakeCommentId: randomUUID(), }, }) .where(eq(heartbeatRuns.id, seeded.successorRunId!)); await db .update(issues) .set({ executionRunId: seeded.successorRunId }) .where(eq(issues.id, seeded.issueId)); const [dedicatedRun] = await db .insert(heartbeatRuns) .values({ id: randomUUID(), companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "native", driverKind: "codex", contextSnapshot: { issueId: seeded.issueId, interactionId: seeded.interaction.id, sourceRunId: seeded.sourceRunId, source: "issue.interaction.respond", }, }) .returning(); if (!dedicatedRun) throw new Error("Expected dedicated continuation run"); const wakeup = vi.fn().mockResolvedValue(dedicatedRun); const cancelRun = vi.fn().mockResolvedValue({ id: seeded.sourceRunId, status: "cancelled", }); const steer = vi.fn(); const resolveNativeQuestion = vi .fn() .mockResolvedValue("queued" as const); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup, cancelRun } as never, steer, resolveNativeQuestion, }).deliver(seeded.interaction.id); expect(resolveNativeQuestion).not.toHaveBeenCalled(); expect(steer).not.toHaveBeenCalled(); expect(cancelRun).toHaveBeenCalledWith( seeded.sourceRunId, "Superseded by a dedicated external-chat answer continuation", expect.objectContaining({ errorCode: "external_chat_continuation", resultJson: expect.objectContaining({ interactionId: seeded.interaction.id, externalChatContinuation: true, }), terminationGraceMs: 2_000, }), ); expect(cancelRun.mock.invocationCallOrder[0]).toBeLessThan( wakeup.mock.invocationCallOrder[0]!, ); expect(wakeup).toHaveBeenCalledWith( seeded.agentId, expect.objectContaining({ allowRunCoalescing: false, idempotencyKey: `question-response:${seeded.interaction.id}`, payload: expect.objectContaining({ externalChatContinuation: true, sourceCommentId: seeded.sourceCommentId, wakeCommentId: seeded.sourceCommentId, wakeCommentIds: [seeded.sourceCommentId], }), contextSnapshot: expect.objectContaining({ externalChatContinuation: true, sourceCommentId: seeded.sourceCommentId, wakeCommentId: seeded.sourceCommentId, wakeCommentIds: [seeded.sourceCommentId], }), }), ); expect(outcome).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: dedicatedRun.id, }); expect(dedicatedRun.id).not.toBe(seeded.sourceRunId); const [delivery] = await db .select() .from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ lastErrorCode: "steering_external_chat_context_incompatible", targetRunId: dedicatedRun.id, }); }, ); it("does not promote a source-run wake comment from another issue", async () => { const seeded = await seed(); const goalId = await db .select({ goalId: issues.goalId }) .from(issues) .where(eq(issues.id, seeded.issueId)) .then((rows) => rows[0]?.goalId); if (!goalId) throw new Error("Expected source issue goal"); const otherIssueId = randomUUID(); const otherCommentId = randomUUID(); await db.insert(issues).values({ id: otherIssueId, companyId: seeded.companyId, goalId, title: "Unrelated issue", status: "todo", priority: "medium", }); await db.insert(issueComments).values({ id: otherCommentId, companyId: seeded.companyId, issueId: otherIssueId, authorUserId: "external-user", body: "This other issue must remain outside the continuation.", }); await db .update(heartbeatRuns) .set({ contextSnapshot: { issueId: seeded.issueId, source: "chat:slack", wakeCommentId: otherCommentId, wakeCommentIds: [otherCommentId], }, }) .where(eq(heartbeatRuns.id, seeded.sourceRunId)); const fallbackRunId = randomUUID(); await db.insert(heartbeatRuns).values({ id: fallbackRunId, companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "native", driverKind: "codex", contextSnapshot: { issueId: seeded.issueId, interactionId: seeded.interaction.id, source: "issue.interaction.respond", }, }); const wakeup = vi.fn().mockResolvedValue({ id: fallbackRunId, driverKind: "codex", }); const resolveNativeQuestion = vi.fn().mockResolvedValue("queued" as const); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, resolveNativeQuestion, }).deliver(seeded.interaction.id); expect(resolveNativeQuestion).not.toHaveBeenCalled(); expect(outcome).toMatchObject({ mode: "wake_fallback", targetRunId: fallbackRunId, }); const wakeOptions = wakeup.mock.calls[0]?.[1]; expect(wakeOptions?.payload).toMatchObject({ sourceCommentId: null }); expect(wakeOptions?.payload).not.toHaveProperty("wakeCommentId"); expect(wakeOptions?.payload).not.toHaveProperty("wakeCommentIds"); expect(wakeOptions?.contextSnapshot).toMatchObject({ sourceCommentId: null, }); expect(wakeOptions?.contextSnapshot).not.toHaveProperty("wakeCommentId"); expect(wakeOptions?.contextSnapshot).not.toHaveProperty("wakeCommentIds"); }); it("resolves an in-flight native input request before creating a continuation", async () => { const seeded = await seed({ adapterType: "paperclip_runner", runtimeMode: "native", sourceStatus: "running", }); const wakeup = vi.fn(); const resolveNativeQuestion = vi.fn().mockResolvedValue("queued" as const); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, resolveNativeQuestion, }).deliver(seeded.interaction.id); expect(outcome).toMatchObject({ status: "delivered", mode: "steered", targetRunId: seeded.sourceRunId, }); expect(resolveNativeQuestion).toHaveBeenCalledWith( expect.objectContaining({ id: seeded.interaction.id, status: "answered", }), ); expect(wakeup).not.toHaveBeenCalled(); }); it("keeps native input delivery pending while its PRP session is unavailable", async () => { const seeded = await seed({ adapterType: "paperclip_runner", runtimeMode: "native", sourceStatus: "running", }); const wakeup = vi.fn(); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, resolveNativeQuestion: vi.fn().mockResolvedValue("pending" as const), }).deliver(seeded.interaction.id); expect(outcome).toBeNull(); expect(wakeup).not.toHaveBeenCalled(); const [delivery] = await db.select().from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ status: "pending", attemptCount: 1, errorCount: 0, lastErrorCode: "native_question_session_unavailable", }); }); it("coalesces into a queued successor without creating another wake", async () => { const seeded = await seed({ successorStatus: "queued" }); const successor = await db .select() .from(heartbeatRuns) .where(eq(heartbeatRuns.id, seeded.successorRunId!)) .then((rows) => rows[0]!); const wakeup = vi.fn().mockResolvedValue(successor); const steer = vi.fn(); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer, }).deliver(seeded.interaction.id); expect(outcome).toMatchObject({ status: "delivered", mode: "coalesced", targetRunId: successor.id, }); expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); expect(wakeup.mock.calls[0]?.[1]).toMatchObject({ idempotencyKey: `question-response:${seeded.interaction.id}`, contextSnapshot: { interactionId: seeded.interaction.id, interactionStatus: "answered", }, }); }); it("never steers into the source run and keeps a skipped wake retryable", async () => { const seeded = await seed({ sourceStatus: "running" }); const wakeup = vi.fn().mockResolvedValue(null); const steer = vi.fn(); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer, }).deliver(seeded.interaction.id); expect(outcome).toBeNull(); expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); expect(JSON.stringify(wakeup.mock.calls[0])).not.toContain("Internal API"); expect(JSON.stringify(wakeup.mock.calls[0])).not.toContain("Node.js"); const [delivery] = await db.select().from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ status: "pending", attemptCount: 1, lastErrorCode: "question_response_wake_skipped", }); }); it("delivers after wake suppression outlasts the bounded error retry limit", async () => { const seeded = await seed({ sourceStatus: "running" }); const fallbackRunId = randomUUID(); let wakeAttempts = 0; const wakeup = vi.fn().mockImplementation(async () => { wakeAttempts += 1; if (wakeAttempts <= 5) return null; return db .insert(heartbeatRuns) .values({ id: fallbackRunId, companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", driverKind: "codex", contextSnapshot: { issueId: seeded.issueId }, }) .returning() .then((rows) => rows[0]!); }); const service = questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer: vi.fn(), }); for (let attempt = 0; attempt < 5; attempt += 1) { await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull(); } const delivered = await service.deliver(seeded.interaction.id); expect(delivered).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: fallbackRunId, }); const [delivery] = await db.select().from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ status: "fallback_queued", attemptCount: 6, errorCount: 0, targetRunId: fallbackRunId, }); }); it("preserves the actual-error retry budget after prolonged wake suppression", async () => { const seeded = await seed({ sourceStatus: "running" }); const fallbackRunId = randomUUID(); let wakeAttempts = 0; const wakeup = vi.fn().mockImplementation(async () => { wakeAttempts += 1; if (wakeAttempts <= 5) return null; if (wakeAttempts === 6) throw new Error("scheduler temporarily unavailable"); return db .insert(heartbeatRuns) .values({ id: fallbackRunId, companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", driverKind: "codex", contextSnapshot: { issueId: seeded.issueId }, }) .returning() .then((rows) => rows[0]!); }); const service = questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer: vi.fn(), }); for (let attempt = 0; attempt < 5; attempt += 1) { await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull(); } await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull(); const [afterError] = await db .select() .from(issueQuestionResponseDeliveries); expect(afterError).toMatchObject({ status: "pending", attemptCount: 6, errorCount: 1, lastErrorCode: "scheduler temporarily unavailable", }); await expect(service.deliver(seeded.interaction.id)).resolves.toMatchObject( { status: "fallback_queued", targetRunId: fallbackRunId, }, ); const [delivered] = await db.select().from(issueQuestionResponseDeliveries); expect(delivered).toMatchObject({ status: "fallback_queued", attemptCount: 7, errorCount: 1, targetRunId: fallbackRunId, }); }); it("enforces one durable wake per question-response idempotency key", async () => { const seeded = await seed({ sourceStatus: "running" }); const idempotencyKey = `question-response:${seeded.interaction.id}`; const request = { companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system", reason: "issue_commented", idempotencyKey, } as const; await db .insert(agentWakeupRequests) .values({ ...request, status: "queued" }); await expect( db.insert(agentWakeupRequests).values({ ...request, status: "coalesced", }), ).rejects.toMatchObject({ cause: { code: "23505" } }); // Suppression receipts are intentionally outside the fence so the outbox // can retry after scheduling is enabled again. await expect( db.insert(agentWakeupRequests).values({ ...request, status: "skipped", finishedAt: new Date(), }), ).resolves.toBeDefined(); }); it("reuses the winning wake when a concurrent insert hits the idempotency fence", async () => { const seeded = await seed({ sourceStatus: "running" }); const fallbackRunId = randomUUID(); const wakeup = vi .fn() .mockImplementation( async ( _agentId: string, options: { idempotencyKey?: string | null }, ) => { const request = { companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system", reason: "issue_commented", idempotencyKey: options.idempotencyKey, } as const; const [winner] = await db .insert(agentWakeupRequests) .values({ ...request, status: "queued", }) .returning(); await db.insert(heartbeatRuns).values({ id: fallbackRunId, companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", driverKind: "codex", wakeupRequestId: winner!.id, contextSnapshot: { issueId: seeded.issueId }, }); await db .update(agentWakeupRequests) .set({ runId: fallbackRunId }) .where(eq(agentWakeupRequests.id, winner!.id)); // Model the losing claimant reaching the same transactional insert after // the winner commits. The service must recover the winner's receipt. await db.insert(agentWakeupRequests).values({ ...request, status: "coalesced", }); throw new Error("unreachable"); }, ); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer: vi.fn(), }).deliver(seeded.interaction.id); expect(outcome).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: fallbackRunId, }); expect(wakeup).toHaveBeenCalledTimes(1); const [delivery] = await db.select().from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ status: "fallback_queued", attemptCount: 1, errorCount: 0, targetRunId: fallbackRunId, }); }); it("settles against a canonical wake from the previous assignee without retrying forever", async () => { const seeded = await seed({ sourceStatus: "running" }); const replacementAgentId = randomUUID(); await db.insert(agents).values({ id: replacementAgentId, companyId: seeded.companyId, name: "Replacement runner", role: "engineer", status: "active", adapterType: "codex_local", adapterConfig: {}, runtimeConfig: {}, permissions: {}, }); await db .update(issues) .set({ assigneeAgentId: replacementAgentId }) .where(eq(issues.id, seeded.issueId)); await db.insert(agentWakeupRequests).values({ companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system", reason: "issue_commented", status: "queued", idempotencyKey: `question-response:${seeded.interaction.id}`, }); const wakeup = vi.fn(); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer: vi.fn(), }).deliver(seeded.interaction.id); expect(wakeup).not.toHaveBeenCalled(); expect(outcome).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: null, }); const [delivery] = await db.select().from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ status: "fallback_queued", attemptCount: 1, errorCount: 0, }); }); it("reuses a durable wake receipt instead of issuing a duplicate continuation", async () => { const seeded = await seed({ sourceStatus: "running" }); const [wakeRequest] = await db .insert(agentWakeupRequests) .values({ companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system", reason: "issue_commented", status: "queued", idempotencyKey: `question-response:${seeded.interaction.id}`, }) .returning(); const [wakeRun] = await db .insert(heartbeatRuns) .values({ companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", wakeupRequestId: wakeRequest!.id, contextSnapshot: { issueId: seeded.issueId }, }) .returning(); await db .update(agentWakeupRequests) .set({ runId: wakeRun!.id }) .where(eq(agentWakeupRequests.id, wakeRequest!.id)); const wakeup = vi.fn(); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer: vi.fn(), }).deliver(seeded.interaction.id); expect(outcome).toMatchObject({ status: "delivered", mode: "coalesced", targetRunId: wakeRun!.id, }); expect(wakeup).not.toHaveBeenCalled(); }); it("recovers a completed wake when receipt finalization was interrupted", async () => { const seeded = await seed({ sourceStatus: "running" }); const [wakeRequest] = await db .insert(agentWakeupRequests) .values({ companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system", reason: "issue_commented", status: "completed", idempotencyKey: `question-response:${seeded.interaction.id}`, finishedAt: new Date(), }) .returning(); const [wakeRun] = await db .insert(heartbeatRuns) .values({ companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "succeeded", runtimeMode: "legacy", wakeupRequestId: wakeRequest!.id, contextSnapshot: { issueId: seeded.issueId }, finishedAt: new Date(), }) .returning(); await db .update(agentWakeupRequests) .set({ runId: wakeRun!.id }) .where(eq(agentWakeupRequests.id, wakeRequest!.id)); const wakeup = vi.fn(); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer: vi.fn(), }).deliver(seeded.interaction.id); expect(outcome).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: wakeRun!.id, }); expect(wakeup).not.toHaveBeenCalled(); const [delivery] = await db.select().from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ status: "fallback_queued", errorCount: 0, targetRunId: wakeRun!.id, }); }); it("keeps a long wake claim leased while the side effect is active", async () => { const seeded = await seed({ sourceStatus: "running" }); // Each call gets its own resolver. A second, unexpected call fails at // once instead of sharing one resolver with the first call and hanging. const wakeResolvers: Array<(value: null) => void> = []; const wakeup = vi.fn(() => { if (wakeResolvers.length > 0) { throw new Error( "wakeup was invoked a second time for the same claim; a re-entrant delivery must fail at once, not hang", ); } return new Promise((resolve) => { wakeResolvers.push(resolve); }); }); // The renewal timer and the sweep both read time from this injected // clock. The test moves the clock by hand, so the assertions below do // not depend on the speed of a database round trip or a timer callback. let clock = new Date("2026-01-01T00:00:00.000Z"); const service = questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer: vi.fn(), now: () => clock, claimStaleMs: 40, claimRefreshMs: 5, }); const deliveryPromise = service.deliver(seeded.interaction.id); await vi.waitFor(() => expect(wakeup).toHaveBeenCalledTimes(1)); const [claimedRow] = await db .select() .from(issueQuestionResponseDeliveries) .where( eq( issueQuestionResponseDeliveries.interactionId, seeded.interaction.id, ), ); const claimedAt = claimedRow!.lastAttemptAt!.getTime(); // Move the clock past claimStaleMs, then wait for a real renewal tick to // pick it up. This proves the renewal ran. It does not only assert that // a fixed real-time sleep was long enough. clock = new Date(clock.getTime() + 1000); await vi.waitFor(async () => { const [row] = await db .select() .from(issueQuestionResponseDeliveries) .where( eq( issueQuestionResponseDeliveries.interactionId, seeded.interaction.id, ), ); expect(row?.lastAttemptAt?.getTime()).toBeGreaterThan(claimedAt); }); await expect(service.sweepPending()).resolves.toMatchObject({ scanned: 0 }); wakeResolvers[0]!(null); await expect(deliveryPromise).resolves.toBeNull(); expect(wakeup).toHaveBeenCalledTimes(1); const [delivery] = await db .select() .from(issueQuestionResponseDeliveries) .where( eq( issueQuestionResponseDeliveries.interactionId, seeded.interaction.id, ), ); expect(delivery).toMatchObject({ status: "pending", attemptCount: 1 }); }); it("fences a stale worker after a newer claim generation takes ownership", async () => { const seeded = await seed({ sourceStatus: "running" }); let releaseFirstWake!: (value: { id: string; driverKind: string }) => void; const firstWakeup = vi.fn( () => new Promise<{ id: string; driverKind: string }>((resolve) => { releaseFirstWake = resolve; }), ); const firstService = questionResponseDeliveryService(db, { heartbeat: { wakeup: firstWakeup } as never, steer: vi.fn(), claimStaleMs: 40, claimRefreshMs: 5, }); const firstDelivery = firstService.deliver(seeded.interaction.id); await vi.waitFor(() => expect(firstWakeup).toHaveBeenCalledTimes(1)); // Simulate recovery after the first worker stopped renewing. The next // claim increments attemptCount, which is the fencing generation. await db .update(issueQuestionResponseDeliveries) .set({ status: "pending", lastAttemptAt: new Date(0), }) .where( eq( issueQuestionResponseDeliveries.interactionId, seeded.interaction.id, ), ); const secondRunId = randomUUID(); const secondWakeup = vi.fn().mockImplementation(async () => db .insert(heartbeatRuns) .values({ id: secondRunId, companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", driverKind: "codex", contextSnapshot: { issueId: seeded.issueId }, }) .returning() .then((rows) => rows[0]!), ); const secondOutcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup: secondWakeup } as never, steer: vi.fn(), }).deliver(seeded.interaction.id); expect(secondOutcome).toMatchObject({ status: "fallback_queued", targetRunId: secondRunId, duplicate: false, }); releaseFirstWake({ id: randomUUID(), driverKind: "codex" }); await expect(firstDelivery).resolves.toMatchObject({ status: "fallback_queued", targetRunId: secondRunId, duplicate: true, }); const [delivery] = await db.select().from(issueQuestionResponseDeliveries); expect(delivery).toMatchObject({ status: "fallback_queued", targetRunId: secondRunId, attemptCount: 2, }); const deliveryEvents = await db .select() .from(activityLog) .where(eq(activityLog.action, "issue.question_response_delivered")); expect(deliveryEvents).toHaveLength(1); }); it.each(DIRECT_ADAPTER_TYPES)( "keeps %s on the existing wake path without invoking native steering", async (adapterType) => { const seeded = await seed({ adapterType, runtimeMode: "legacy", successorStatus: "running", }); const fallbackRunId = randomUUID(); const wakeup = vi.fn().mockImplementation(async () => db .insert(heartbeatRuns) .values({ id: fallbackRunId, companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", contextSnapshot: { issueId: seeded.issueId }, }) .returning() .then((rows) => rows[0]!), ); const steer = vi.fn(); const outcome = await questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer, }).deliver(seeded.interaction.id); expect(outcome).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: fallbackRunId, }); expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); expect(wakeup.mock.calls[0]?.[1]).toMatchObject({ idempotencyKey: `question-response:${seeded.interaction.id}`, contextSnapshot: { issueId: seeded.issueId, interactionId: seeded.interaction.id, interactionStatus: "answered", }, }); }, ); it("queues once without probing successor steering", async () => { const seeded = await seed({ successorStatus: "running" }); const fallbackRunId = randomUUID(); const steer = vi.fn().mockRejectedValue( Object.assign(new Error("unsupported"), { code: "steering_unsupported", }), ); const wakeup = vi.fn().mockImplementation(async () => db .insert(heartbeatRuns) .values({ id: fallbackRunId, companyId: seeded.companyId, agentId: seeded.agentId, invocationSource: "automation", status: "queued", driverKind: "codex", contextSnapshot: { issueId: seeded.issueId }, }) .returning() .then((rows) => rows[0]!), ); const service = questionResponseDeliveryService(db, { heartbeat: { wakeup } as never, steer, }); const first = await service.deliver(seeded.interaction.id); const second = await service.deliver(seeded.interaction.id); expect(first).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: fallbackRunId, }); expect(second?.duplicate).toBe(true); expect(steer).not.toHaveBeenCalled(); expect(wakeup).toHaveBeenCalledTimes(1); }); it("formats text, select labels, multi-select, and custom answers in order", async () => { const seeded = await seed(); const envelope = buildQuestionResponseDeliveryEnvelope(seeded.interaction); expect(envelope.response).toEqual({ schema: "paperclip.question_response.v1", answers: { purpose: { text: "Internal API" }, runtime: { selectedOptionIds: ["node"] }, features: { selectedOptionIds: ["health", "logs"], customText: "Metrics", }, }, }); expect(formatQuestionResponseSteeringMessage(envelope)).toBe( [ "Answered questions", "", "- Purpose — What is it for?: Internal API", "- Runtime — Which runtime?: Node.js", "- Features — Which features?: Health check, Request logs, Metrics", ].join("\n"), ); }); });