From 78adeadc4b99c396e9bbbd8a3c35125a445ba751 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 07:49:47 -0500 Subject: [PATCH] fix(server): signal chat milestones before queue cancellation writes Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 9 +++ ...26-10-25-chat-run-milestone-sleep-audit.md | 45 +++++++++++++ .../__tests__/chat-run-milestone-work.test.ts | 66 ++++++++++++++++++- server/src/services/heartbeat/queue.ts | 17 ++++- 4 files changed, 133 insertions(+), 4 deletions(-) diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 0240ecbfdf9..e44424c7d7f 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1453,3 +1453,12 @@ fixture cleanup. The audit preserves setup failures and provider-acceptance limits. No periodic projector has been removed yet. Fresh remote inspection still finds #17590 Paperclip Review queued, #17592 neutral, and #17490 E2E shard 3 plus aggregate failing. Those gates and all remaining original scope stay open. + + +Slice 28 raw queue cancellation checkpoint: general chat-control recovery, +superseded case handoffs, stale scheduled retries and reassigned execution owners +now signal before terminal writes. Missing notices were reproduced for the first +two; replacement-admission tests cover the latter two and their early intent +placement is audited. All 60 milestone/queue tests and server production types +pass. Queued-comment/lineage and remaining dependency producers still prevent +removing periodic projection. No new PR, merge, or full-goal completion is claimed. diff --git a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md index bea32e9040b..ab8b25f680b 100644 --- a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md @@ -483,3 +483,48 @@ Evidence and limitations: Raw queue cancellation, queued-comment/lineage changes, native reviewer claim acceptance, and the remaining dependency producer audit still prevent retiring the production periodic projector. All original stack scope remains required. + + +## Remaining queue cancellation writers + +The general chat-control recovery gate now registers intent before its guarded +terminal write, after the existing proof/no-op decisions. Invalid admission at +dispatch still fails closed and never calls the provider callback. Repeated +attempts against a settled run exit before signaling. + +Superseded case handoffs previously cancelled with a root-client UPDATE. That +single status change now owns a narrow transaction, locks/selects the still-queued +run, and registers company-scoped intent before updating it. The company scope +avoids trusting a superseded handoff's issue identity as the run's current chat +binding. Status/company guards and the existing post-update wake-status callback +remain unchanged. A repeated claim finds no queued row and creates no intent. +The added run-first lock does not enclose provider I/O or introduce an issue lock. + +Admission's stale scheduled-retry and reassigned execution-owner cancellations +also signal before their existing transaction writes. Those transactions may +later insert/notify a replacement run, but that later intent is insufficient for +the protocol's before-first-write requirement. Company/issue identity comes from +the owning admission state; no new module port or provider branch is added. + +Evidence: + +- `/tmp/sleep28-case-handoff-before.log` reproduces committed cancellation with + no terminal notice before the fix. +- `/tmp/sleep28-chat-control-before.log` reproduces the same gap for a failed + recovery gate; the repaired case-handoff regression passes in that run. +- Both gate tests verify no provider execution and quiet repeated attempts. +- The two real replacement-admission cases cover queued and scheduled-retry + predecessors, their distinct cancellation codes, the old run's terminal notice, + and the drain-held replacement. These are behavioral coverage, not claimed + missing-notice-before reproductions: later admission signaling can also wake + the same issue. Their early intent placement is established by code audit. +- `/tmp/sleep28-cancellation-final-tests.log`: all 60 milestone/queue cases pass, + including 35 milestone worker cases. Earlier 58-case and two-case runs overlap. +- `/tmp/sleep28-cancellation-final-typecheck.log`: server production types pass. + Test files are excluded. `git diff --check` passes. No PR acceptance claimed. + +The identified raw queue cancellation and native wake receipt writers are now +covered. Still required: queued-comment/lineage mutations, native reviewer claim +acceptance, publication and interaction changes, conversation/configuration, +question claim/retry, and monitor/semantic receipt dependencies. Production still +uses the combined periodic projector until those producers are complete. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 4ca86eda572..76199630d96 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-work.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, agentWakeupRequests, authUsers, companyMemberships, chatConversations, chatEndpoints, chatMessageLinks, chatPublications, companies, completionContracts, createDb, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, workAssessments, subscribeDatabaseWork, +import { activityLog, agents, agentWakeupRequests, authUsers, caseHandoffs, cases, companyMemberships, chatConversations, chatEndpoints, chatMessageLinks, chatPublications, companies, completionContracts, createDb, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, workAssessments, subscribeDatabaseWork, heartbeatRunEvents, heartbeatRuns, issueComments, issueQuestionResponseDeliveries, issueThreadInteractions, issues, toolApplications, toolConnections } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; @@ -15,6 +15,7 @@ import { createAdmissionTransactionScope, createPostgresWakeQueueAdapter, create import { instanceSettingsService } from "../services/instance-settings.js"; import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js"; +import { CHAT_CONTROL_RECOVERY_ADMISSION_KEY, CHAT_CONTROL_RECOVERY_UNRESOLVED_CODE } from "../services/chat-control-recovery-stop.js"; import { heartbeatService } from "../services/heartbeat.js"; import { commitNativeStatusDecision, dispatchPendingNativeStatusEffects } from "../services/native-runtime/status-decision-committer.js"; import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; @@ -65,6 +66,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); for (const companyId of fixtureCompanies.splice(0)) { await db.delete(activityLog).where(eq(activityLog.companyId, companyId)); + await db.delete(caseHandoffs).where(eq(caseHandoffs.companyId, companyId)); + await db.delete(cases).where(eq(cases.companyId, companyId)); await db.delete(chatMessageLinks).where(eq(chatMessageLinks.companyId, companyId)); await db.delete(chatPublications).where(eq(chatPublications.companyId, companyId)); await db.delete(chatConversations).where(eq(chatConversations.companyId, companyId)); @@ -248,6 +251,67 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(await service.claimQueuedRun(before)).toBeNull(); await quiet(); }); + it("wakes when claim rejects a superseded case handoff and leaves repeated claims quiet", async () => { + const wakeId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: wakeId, companyId: f.companyId, agentId: f.agentId, + source: "assignment", status: "queued", payload: { issueId: f.issueId }, + }); + const seeded = await run(f, { status: "queued", runtimeMode: "legacy", wakeupRequestId: wakeId }); + const [record] = await db.insert(cases).values({ companyId: f.companyId, caseNumber: 1, identifier: "M-CASE-1", + caseType: "work", title: "Chat task", workIssueId: f.issueId, handoffVersion: 2 }).returning(); + await db.insert(caseHandoffs).values({ companyId: f.companyId, caseId: record.id, issueId: f.issueId, + version: 1, toAgentId: f.agentId, phase: "implementation", instructions: "Build the change", actorType: "system", + actorId: "test", idempotencyKey: "superseded", inputFingerprint: "fixture", wakeupRequestId: wakeId, + expectedIssueStatusVersion: 0, appliedIssueStatusVersion: 0, state: "delivered" }); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.id)); + const { service, dependencies } = await queue(); + await start(); await flush(); await quiet(); + expect(await service.claimQueuedRun(before)).toBeNull(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.id)))[0]) + .toMatchObject({ status: "cancelled", errorCode: "case_handoff_superseded" }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${seeded.id}:failed:${f.endpointId}`); + expect(dependencies.executeRun).not.toHaveBeenCalled(); + await quiet(); + expect(await service.claimQueuedRun(before)).toBeNull(); + await quiet(); + }); + it("wakes when dispatch fails an unverifiable chat recovery and leaves repeated gates quiet", async () => { + const seeded = await run(f, { status: "running", runtimeMode: "legacy", invocationSource: "automation", + runnerProfileJson: { [CHAT_CONTROL_RECOVERY_ADMISSION_KEY]: {} }, + }); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.id)); + const { service, dependencies } = await queue(); + const dispatch = vi.fn(async () => null); + await start(); await flush(); await quiet(); + await service.withChatControlRecoveryGate(before, "dispatch", dispatch); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.id)))[0]) + .toMatchObject({ status: "failed", errorCode: CHAT_CONTROL_RECOVERY_UNRESOLVED_CODE }); + expect(dispatch).not.toHaveBeenCalled(); expect(dependencies.executeRun).not.toHaveBeenCalled(); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${seeded.id}:failed:${f.endpointId}`); + await quiet(); + await service.withChatControlRecoveryGate(before, "dispatch", dispatch); + await quiet(); + }); + it.each(["queued", "scheduled_retry"] as const)("publishes cancellation when admission replaces a reassigned %s owner", async status => { + const old = await run(f, { status, runtimeMode: "legacy", scheduledRetryAt: status === "scheduled_retry" ? new Date(Date.now() + 60_000) : null }); + const [replacement] = await db.insert(agents).values({ companyId: f.companyId, name: "Replacement", adapterType: "paperclip_runner" }).returning(); + await db.update(issues).set({ assigneeAgentId: replacement.id, executionRunId: old.id }).where(eq(issues.id, f.issueId)); + const { service, dependencies } = await queue(); + dependencies.getAgent.mockResolvedValue(replacement); + dependencies.getSchedulingSuppression.mockResolvedValue({ suppressed: true, reason: "task_drain" }); + await start(); await flush(); await quiet(); + const admitted = await service.enqueueWakeup(replacement.id, { source: "on_demand", reason: "issue_commented", + contextSnapshot: runValues().contextSnapshot, requestedByActorType: "user", requestedByActorId: f.userId, + }); + expect(admitted).toMatchObject({ agentId: replacement.id, status: "queued" }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, old.id)))[0]) + .toMatchObject({ status: "cancelled", errorCode: status === "queued" ? "lock_released_on_reassignment" : "issue_reassigned" }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${old.id}:failed:${f.endpointId}`); + expect(dependencies.executeRun).not.toHaveBeenCalled(); await quiet(); + }); it("wakes when the real coalescing writer gives an existing run chat lineage", async () => { const seeded = await run(f, { contextSnapshot: { issueId: f.issueId, source: "issue_assigned" } }); await start(); await flush(); expect(await publications()).toEqual([]); await quiet(); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index cc499b39df6..68532b79765 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -844,6 +844,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const error = stopped ? "Automatic continuation stopped by the committed chat conversation close. Send a new request in chat or on the Board to start fresh work." : "Automatic continuation source could not be verified before provider admission. Review the task and send a fresh request; this attempt will not automatically retry."; + await notifyChatRunMilestones(tx, { companyId: current.companyId, issueId }); [terminal] = await tx .update(heartbeatRuns) .set({ @@ -959,9 +960,17 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (!["pending_dispatch", "delivered"].includes(caseHandoffReceipt.state) || caseHandoffReceipt.toAgentId !== run.agentId || !current || current.version !== caseHandoffReceipt.version || current.issueId !== caseHandoffReceipt.issueId || current.agentId !== run.agentId || (!run.startedAt && current.statusVersion !== caseHandoffReceipt.appliedIssueStatusVersion)) { - const [cancelled] = await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), - errorCode: "case_handoff_superseded", error: "This case handoff was cancelled or superseded.", updatedAt: new Date() }) - .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "queued"))).returning(); + const cancelled = await db.transaction(async tx => { + const [currentRun] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "queued"))) + .for("update"); + if (!currentRun) return null; + await notifyChatRunMilestones(tx, { companyId: run.companyId }); + const [row] = await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), + errorCode: "case_handoff_superseded", error: "This case handoff was cancelled or superseded.", updatedAt: new Date() }) + .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "queued"))).returning(); + return row; + }); if (cancelled) await setWakeupStatus(run.wakeupRequestId, "cancelled", { finishedAt: new Date(), error: cancelled.error }); return null; } @@ -3417,6 +3426,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const reason = issueCancelled ? "Cancelled because the issue was cancelled before the scheduled retry became due" : "Cancelled because the issue was reassigned before the scheduled retry became due"; + await notifyChatRunMilestones(tx, { companyId: scheduledRun.companyId, issueId: issue.id }); const cancelled = await tx .update(heartbeatRuns) .set({ @@ -3553,6 +3563,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende issue.assigneeAgentId && activeExecutionRun.agentId !== issue.assigneeAgentId ) { + await notifyChatRunMilestones(tx, { companyId: activeExecutionRun.companyId, issueId: issue.id }); const cancelled = await tx .update(heartbeatRuns) .set({