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 2e087b418de..4d1631825e4 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1421,3 +1421,14 @@ worker cases, nine native chat recovery integrations, and server production type pass after repair. Real late continuation receipts now reactivate projection. Next is heartbeat queue/admission/recovery producer coverage. Periodic projection and all remaining inventory/remote acceptance duties remain outstanding. + + +Slice 28 queue checkpoint: issue admission, ordinary/comment claims, drain +release, coalesced lineage and native review claims now notify from their owning +transactions. Real missing-wake regressions were reproduced for admission, +coalescence and all ordinary/comment claim/release paths. The 24 milestone worker +cases and 92 existing queue/policy cases pass, as does server production typecheck. +The audit records the initial fixture FK failure, exact logs, remaining raw +cancellation/promotion/receipt writers, and native reviewer acceptance gap. +Periodic projection remains until producer coverage is complete; no new PR 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 7c4d06adc01..49180b32c28 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 @@ -331,3 +331,69 @@ Remaining interaction, publication, conversation/configuration, question claim/retry, and monitor/semantic-receipt dependencies still require coverage. The new worker is still not production-registered, and the periodic combined projector remains until that audit and its real producer tests are complete. + + +## Queue admission, claim, release and coalescence checkpoint + +The issue-bound admission transaction signals before inserting its run. Ordinary, +legacy-comment and queued-comment claims signal before changing run/wake state. +Queued-comment discard and committed chat-close cancellation branches also signal +before their terminal writes. The native review claim owner signals after its +review/ownership checks and before changing the run to running. No claim inspection +or admission-marker-only update gets a signal merely because it was inspected. + +Drain release now selects and locks the still-running candidate in its existing +transaction, obtains its scope, then registers intent before releasing the run, +wake and issue lock. A repeated release after the run is queued returns without +an intent or update. The added selection follows the same run-first locking order +as the previous UPDATE; it is not a periodic discovery query. + +The wake-queue PostgreSQL admission adapter owns coalesced run-context writes and +signals from that existing transaction. Its port carries company identity, so +this writer uses company-scoped reconstruction rather than adding feature fields +to the module contract or interpreting chat/provider payloads. The actual +conversation/lineage decision remains in the chat projection owner. + +The fallback company-run admission/coalescing branch of `heartbeat/queue.ts` is +outside the issue-bound path and cannot create a chat milestone. It intentionally +has no new notification or wrapper. Any later task-binding writer still needs +its own signal. This is separate from claiming a historically bound run, for +which a missing context issue ID conservatively uses company scope. + +Verification and failures: + +- Initial claim fixtures incorrectly inserted a run before its referenced wake. + Three failed on the real FK, while the drain-release missing-wake regression + failed at its publication assertion. `/tmp/sleep28-queue-claim-before.log`. + Fixing insertion and cleanup dependency order required no production change. +- With valid fixtures, all four claim/release regressions failed at the missing + notice assertion before wiring: `/tmp/sleep28-queue-claim-corrected-before.log`. + After repair all 22 then-current worker cases passed: + `/tmp/sleep28-queue-claim-tests.log`. +- Real issue admission during a held execution drain and the actual PostgreSQL + coalescing writer each reproduced a missing notice independently before their + notification was added: `/tmp/sleep28-queue-admission-before.log` and + `/tmp/sleep28-coalescing-before.log`. +- All 24 worker cases now pass: `/tmp/sleep28-queue-producers-tests.log`. Queue + tests call the actual admission/claim/release implementations and database; + dependency callbacks hold execution and prevent provider dispatch. The + coalescing case exercises the actual adapter inside an outer transaction. + These do not claim provider execution or a full native reviewer admission test. +- All 92 existing queue-module and wake-admission policy tests pass: + `/tmp/sleep28-queue-module-tests.log`. The queue module now accepts the standard + guarded external test URL through `PAPERCLIP_QUEUE_TEST_DATABASE_URL`; its + destructive per-case fixture reset ran only on the new isolated database + `paperclip_sleep28_queue_claim_20261025_test`. Embedded tests remain the default. +- Server production typecheck passes: `/tmp/sleep28-queue-producers-typecheck.log`. + This excludes test files and is not workspace/PR acceptance. + +Remaining queue producer audit is explicit: the general chat-control recovery +gate's terminal write, early case-handoff cancellation, stale retry/reassignment +cancellation, native status-wake dispatch receipts, deferred-run promotion in the +wake-queue adapter, queued-comment mutations and run-dispatch cancellation. The +native review claim signal also needs its dedicated real admission acceptance +case before final slice acceptance. Other recovery, interaction, presentation, +conversation/configuration, question claim/retry and monitor/semantic receipt +writers from the dependency table remain. Production still uses the periodic +combined projector; the new worker is not registered yet. No original inventory +or PR-stack acceptance requirement is removed by this checkpoint. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 284a4c80f37..24fbc433482 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-work.test.ts @@ -10,6 +10,10 @@ import { chatRunMilestoneWork } from "../services/chat-run-milestone-work.js"; import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { questionResponseDeliveryService } from "../services/question-response-delivery.js"; +import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.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"; @@ -72,9 +76,9 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.delete(statusDecisionEffects).where(eq(statusDecisionEffects.companyId, companyId)); await db.delete(statusDecisions).where(eq(statusDecisions.companyId, companyId)); await db.delete(workAssessments).where(eq(workAssessments.companyId, companyId)); - await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); await db.delete(nativeRunResults).where(eq(nativeRunResults.companyId, companyId)); await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); await db.delete(completionContracts).where(eq(completionContracts.companyId, companyId)); await db.delete(issues).where(eq(issues.companyId, companyId)); await db.delete(agents).where(eq(agents.companyId, companyId)); @@ -137,6 +141,75 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu ].sort()); await quiet(); }); + async function queue() { + const dependencies = createHeartbeatQueueTestDependencies(db); + const [agent] = await db.select().from(agents).where(eq(agents.id, f.agentId)); + dependencies.getAgent.mockResolvedValue(agent); + dependencies.instanceSettings.getExperimental.mockImplementation(() => instanceSettingsService(db).getExperimental()); + return { service: createHeartbeatQueue(db, dependencies), dependencies }; + } + it("wakes on real queue admission while execution is held by drain", async () => { + const { service, dependencies } = await queue(); + dependencies.getSchedulingSuppression.mockResolvedValue({ suppressed: true, reason: "task_drain" }); + await start(); await flush(); await quiet(); + const admitted = await service.enqueueWakeup(f.agentId, { source: "on_demand", reason: "issue_commented", + contextSnapshot: runValues().contextSnapshot, requestedByActorType: "user", requestedByActorId: f.userId, + }); + expect(admitted).toMatchObject({ status: "queued", companyId: f.companyId }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${admitted!.id}:queued:${f.endpointId}`); + expect(dependencies.executeRun).not.toHaveBeenCalled(); + await quiet(); + }); + it.each(["ordinary", "legacy_comments", "queued_comments"] as const)("wakes on a real %s queue claim", async kind => { + const wakeId = randomUUID(); + const context = runValues().contextSnapshot; + await db.insert(agentWakeupRequests).values({ id: wakeId, companyId: f.companyId, agentId: f.agentId, + source: "on_demand", status: "queued", + payload: kind === "queued_comments" ? { _paperclipWakeContext: { wakeCommentIds: [f.commentId] } } : {}, + }); + const seeded = await run(f, { status: "queued", runtimeMode: "legacy", wakeupRequestId: wakeId, + contextSnapshot: kind === "ordinary" ? context : { ...context, wakeCommentIds: [f.commentId] }, + }); + await db.update(agentWakeupRequests).set({ runId: seeded.id }).where(eq(agentWakeupRequests.id, wakeId)); + 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)).toMatchObject({ id: seeded.id, status: "running" }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${seeded.id}:working:${f.endpointId}`); + expect(dependencies.executeRun).not.toHaveBeenCalled(); + await quiet(); + expect(await service.claimQueuedRun(before)).toBeNull(); + 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(); + await db.transaction(async tx => { + await createWakeAdmissionWriter().coalesceIntoActiveExecutionRun( + createAdmissionTransactionScope(f.companyId, tx as unknown as typeof db), { + companyId: f.companyId, activeExecutionRunId: seeded.id, mergedContextSnapshot: runValues().contextSnapshot, + agentId: f.agentId, source: "on_demand", triggerDetail: null, payload: { issueId: f.issueId }, + requestedByActorType: "user", requestedByActorId: f.userId, idempotencyKey: null, + }); + expect(await publications()).toEqual([]); + }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${seeded.id}:working:${f.endpointId}`); + await quiet(); + }); + it("wakes on a claim released for drain and leaves repeated release quiet", async () => { + const seeded = await run(); + const { service } = await queue(); + await start(); await flush(); await quiet(); + await service.releaseRunClaimedJustBeforeSuppression(seeded.id); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${seeded.id}:queued:${f.endpointId}`); + await quiet(); + await service.releaseRunClaimedJustBeforeSuppression(seeded.id); + await quiet(); + }); async function nativeFixture(phase = "committed") { const contractId = randomUUID(), resultId = randomUUID(); await db.insert(completionContracts).values({ diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index c4929f482ba..44717127131 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; import { readRunCancellation } from "../../../services/run-cancellation.js"; @@ -948,6 +949,7 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { async coalesceIntoActiveExecutionRun(scope, input) { const tx = requireAdmissionTx(scope, input.companyId); const now = new Date(); + await notifyChatRunMilestones(tx, { companyId: input.companyId }); const mergedRun = await tx .update(heartbeatRuns) .set({ contextSnapshot: input.mergedContextSnapshot, updatedAt: now }) diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 9af4a5ac956..51e68fbad29 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -3,6 +3,7 @@ import type { Db } from "@paperclipai/db"; import { agents, agentWakeupRequests, caseHandoffs, cases, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; @@ -72,14 +73,15 @@ describe("heartbeat queue module boundary", () => { }); }); -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_QUEUE_TEST_DATABASE_URL"); +const support = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); if (!support.supported) console.warn(`Skipping queue module database tests: ${support.reason}`); describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { - let database: Awaited>; + let database: Awaited> | undefined; let db: Db; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-queue-module"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-queue-module"); + db = createDb(external ?? database!.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end(); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index c6ac63b6a53..dac88a943bb 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { authorizationService } from "../authorization.js"; import { assertTaskRetryConversationAllowed, readTaskRetryIntent, retireSupersededTaskRetry } from "../task-retry.js"; import { insertIssueComment } from "../issue-comment-writes.js"; @@ -1315,6 +1316,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (proof.kind === "stopped") { const error = "Automatic continuation stopped by the committed chat conversation close. Send a fresh request in chat or on the Board."; + await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [cancelled] = await tx .update(heartbeatRuns) .set({ @@ -1393,6 +1395,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // explicitly bound queued-message envelope is subject to the // live-comment discard gate below. await withExternalAdmissionGuard(tx, run.companyId, run.agentId, () => assertNoExternalOverlap(tx, run.companyId, run.agentId)); + await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [claimedRun] = await tx .update(heartbeatRuns) .set({ @@ -1434,6 +1437,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }); if (liveIds.length === 0) { const reason = "Queued messages were discarded before dispatch"; + await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [cancelled] = await tx .update(heartbeatRuns) .set({ @@ -1481,6 +1485,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } await withExternalAdmissionGuard(tx, run.companyId, run.agentId, () => assertNoExternalOverlap(tx, run.companyId, run.agentId)); + await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); await tx .update(agentWakeupRequests) .set({ @@ -1591,6 +1596,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende )).for("update"); if (!lockedRun || lockedRun.status !== "queued") return null; await withExternalAdmissionGuard(claimTx, run.companyId, run.agentId, () => assertNoExternalOverlap(claimTx, run.companyId, run.agentId)); + await notifyChatRunMilestones(claimTx, { companyId: run.companyId, issueId: issueId ?? undefined }); const claimedRun = await claimTx.update(heartbeatRuns).set(claimValues).where(and( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "queued"), )).returning().then((rows) => rows[0] ?? null); @@ -1688,6 +1694,11 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende async function releaseRunClaimedJustBeforeSuppression(runId: string) { const now = new Date(); await db.transaction(async (tx) => { + const [candidate] = await tx.select({ companyId: heartbeatRuns.companyId, contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.status, "running"))).for("update"); + if (!candidate) return; + await notifyChatRunMilestones(tx, { companyId: candidate.companyId, + issueId: readNonEmptyString(parseObject(candidate.contextSnapshot).issueId) ?? undefined }); const released = await tx .update(heartbeatRuns) .set({ @@ -4255,6 +4266,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende adoptedCommentIds = await undeliveredLegacyUserCommentIds(tx as unknown as Db, agent.companyId, issueId, agentId, adoptedCommentIds); } + await notifyChatRunMilestones(tx, { companyId: agent.companyId, issueId: issueId ?? undefined }); const newRun = await tx .insert(heartbeatRuns) .values({ diff --git a/server/src/services/native-runtime/native-review-dispatch.ts b/server/src/services/native-runtime/native-review-dispatch.ts index f51389d53e8..76799a7ebbd 100644 --- a/server/src/services/native-runtime/native-review-dispatch.ts +++ b/server/src/services/native-runtime/native-review-dispatch.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { withExternalAdmissionGuard, assertNoExternalOverlap } from "../../modules/external-agents/index.js"; import { agentWakeupRequests, heartbeatRuns, issues, type Db } from "@paperclipai/db"; import { and, eq } from "drizzle-orm"; @@ -40,6 +41,7 @@ export async function claimQueuedNativeReviewRun(db: Db, input: { contextSnapshot: context, agentNameKey: input.agentNameKey, claimedAt: input.claimedAt, }); if (!locked) return null; + await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); const [claimed] = await tx.update(heartbeatRuns).set({ ...input.claimValues, status: "running", startedAt: run.startedAt ?? input.claimedAt, updatedAt: input.claimedAt,