From 1e72290f03f3884d6cd635217492b601cec464d3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 07:57:00 -0500 Subject: [PATCH] fix(server): signal chat milestones when queued comment lineage changes Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 10 +++ ...26-10-25-chat-run-milestone-sleep-audit.md | 51 +++++++++++++ .../__tests__/chat-run-milestone-work.test.ts | 75 +++++++++++++++++++ .../adapters/queued-comment-postgres.test.ts | 9 ++- .../adapters/queued-comment-postgres.ts | 8 +- 5 files changed, 148 insertions(+), 5 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 e44424c7d7f..1b4a40bab7a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1462,3 +1462,13 @@ 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. + + +Slice 28 lineage checkpoint: queued-comment run-context rewrites, deletion and +cancellation signal through their existing adapter transaction. Real queue +reconstruction and caller-owned discard regressions failed before repair. The +native reviewer claim now has real database integration and causal signal-removal +coverage. All 74 selected worker/adapter/policy cases and server production types +pass. Remaining publication, interaction, configuration, question and monitor +producer coverage still gates periodic-projector removal. Original inventory, +remote gates/findings and final master reconciliation remain open. 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 ab8b25f680b..78012431f27 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 @@ -528,3 +528,54 @@ 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. + + +## Queued-comment lineage and native reviewer claim coverage + +The queued-comment PostgreSQL adapter now signals before rewriting a run's +canonical comment IDs, deleting a queued comment, or cancelling its emptied +queued run. Its private transaction builder receives the already-locked issue +identity; exported application ports and caller wrappers are unchanged. Body-only +edits, issue timestamp touches, and wake-payload bookkeeping do not independently +signal: milestone binding reads comment identities/links and run lineage, not the +queued message text. The same transaction's run rewrite owns the wake for reorder +or nonempty discard; deletion owns the first intent for discard. + +The reconstruction regression deliberately starts with the saved queue and run +context out of sync. The real reorder use case reprojects the run context from +that authoritative queue. It is recovery/compatibility coverage, not a claim that +ordinary reordering introduces new comment identities. The discard regression +uses the actual use case and adapter inside a caller-owned transaction and proves +that intent settles only at outer commit. Existing adapter tests retain rollback +and company/issue authorization coverage. Reference/external-object callbacks in +these fixtures are no-ops; this does not claim their integration behavior. + +The previously outstanding native-review claim test now exercises the real +PostgreSQL claim against a seeded valid review assignment: applied decision, +matching projected status version, pending confirmation with a separate reviewer, +source native run, queued reviewer run and matching wake. It checks issue/run/wake +ownership and the working notice, then verifies repeated claim is quiet. This +covers claim/notification integration, not the entire review-creation workflow. + +Evidence: + +- `/tmp/sleep28-queued-comment-before.log`: reconstructed lineage failed to wake + its notice; discard emitted no transaction intent. Both failed before repair. +- `/tmp/sleep28-native-review-claim-without-signal.log`: removing the real claim + notification reproduced the missing working notice. The signal was restored + in finally; no production native-claim edit remains in this checkpoint. +- `/tmp/sleep28-lineage-final-tests.log`: all 74 selected cases pass across the + 38 milestone worker tests, queued-comment adapter/use cases, and native review + participant policy. Earlier 51-case and targeted native-claim runs overlap. +- `/tmp/sleep28-queued-comment-typecheck.log`: server production types pass; + test files are outside that command. `git diff --check` passes. +- The existing queued-comment adapter tests now accept guarded + `PAPERCLIP_QUEUED_COMMENT_TEST_DATABASE_URL` and close their client. Their broad + fixture cleanup ran only on freshly created/migrated + `paperclip_sleep28_queued_comment_20261025_test`; embedded mode remains default. + +The queued-comment producer and native reviewer claim gaps are closed for this +slice. Publication/interaction, conversation/configuration, question claim/retry, +and monitor/semantic receipt producers still need their own audit and proof. +The production periodic projector remains; no slice publication or full-goal +completion is claimed. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 76199630d96..84f9cf7c634 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-work.test.ts @@ -10,6 +10,8 @@ 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 { createQueuedCommentIssueLockWriter } from "../modules/wake-queue/adapters/queued-comment-postgres.js"; +import { createDiscardQueuedComment, createReorderQueuedComments } from "../modules/wake-queue/application/queued-comment-use-cases.js"; import { createRunDispatch } from "../modules/run-dispatch/index.js"; import { createAdmissionTransactionScope, createPostgresWakeQueueAdapter, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; import { instanceSettingsService } from "../services/instance-settings.js"; @@ -18,6 +20,7 @@ import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue. 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 { claimQueuedNativeReviewRun } from "../services/native-runtime/native-review-dispatch.js"; import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -312,6 +315,49 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${old.id}:failed:${f.endpointId}`); expect(dependencies.executeRun).not.toHaveBeenCalled(); await quiet(); }); + async function editableQueue(staleContext = false) { + const wakeId = randomUUID(); + await db.update(issueComments).set({ authorUserId: f.userId }).where(eq(issueComments.id, f.commentId)); + await db.insert(agentWakeupRequests).values({ id: wakeId, companyId: f.companyId, agentId: f.agentId, + source: "on_demand", status: "queued", payload: { issueId: f.issueId, _paperclipWakeContext: { wakeCommentIds: [f.commentId] } }, + }); + const seeded = await run(f, { status: "queued", runtimeMode: "legacy", wakeupRequestId: wakeId, + contextSnapshot: { ...runValues().contextSnapshot, wakeCommentId: staleContext ? randomUUID() : f.commentId }, + }); + await db.update(agentWakeupRequests).set({ runId: seeded.id }).where(eq(agentWakeupRequests.id, wakeId)); + const issue = { id: f.issueId, companyId: f.companyId, assigneeAgentId: f.agentId, executionRunId: seeded.id }; + const actor = { actorType: "user" as const, actorId: f.userId, agentId: null, runId: null, agentApiKeyId: null }; + const issueLock = createQueuedCommentIssueLockWriter(db, { + syncCommentReferences: async () => {}, deleteCommentReferenceSource: async () => {}, syncCommentExternalObjectsSafely: async () => {}, + }); + const snapshot = await issueLock.withLockedQueue({ issue, actor, queueId: wakeId }, async locked => locked.queue); + return { seeded, issueLock, input: { issue, actor, queueId: wakeId, revision: snapshot.revision, now: new Date() } }; + } + it("wakes when queued-comment mutation reconstructs run lineage from its authoritative queue", async () => { + const { seeded, issueLock, input } = await editableQueue(true); + await start(); await flush(); expect(await publications()).toEqual([]); await quiet(); + await createReorderQueuedComments({ issueLock })({ ...input, orderedCommentIds: [f.commentId] }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${seeded.id}:queued:${f.endpointId}`); + await quiet(); + }); + it("settles queued-comment discard intent only after the owning transaction commits", async () => { + const { seeded, input } = await editableQueue(); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, "chat-run-milestones", event => events.push(event)); + try { + await db.transaction(async tx => { + const writer = createQueuedCommentIssueLockWriter(tx as unknown as typeof db, { + syncCommentReferences: async () => {}, deleteCommentReferenceSource: async () => {}, syncCommentExternalObjectsSafely: async () => {}, + }); + await createDiscardQueuedComment({ issueLock: writer })({ ...input, commentId: f.commentId }); + expect(events).toContain("intent"); expect(events).not.toContain("settled"); + }); + expect(events).toContain("settled"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.id)))[0]) + .toMatchObject({ status: "cancelled", errorCode: "queued_comment_discarded" }); + } finally { unsubscribe(); } + }); 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(); @@ -380,6 +426,35 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu }, }; } + it("wakes from a governed native reviewer claim and leaves repeated claims quiet", async () => { + const input = await continuationDecisionFixture(); + const { decision } = await commitNativeStatusDecision({ db, ...input }); + const [reviewer] = await db.insert(agents).values({ companyId: f.companyId, name: "Reviewer", adapterType: "paperclip_runner" }).returning(); + await db.update(statusDecisions).set({ decisionJson: { ...decision.decisionJson, projectedStatusVersion: 1 } }).where(eq(statusDecisions.id, decision.id)); + await db.update(issues).set({ status: "in_review", statusVersion: 1, lastStatusDecisionId: decision.id, executionRunId: null }).where(eq(issues.id, f.issueId)); + await db.update(chatEndpoints).set({ assignedAgentId: reviewer.id }).where(eq(chatEndpoints.id, f.endpointId)); + const [interaction] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + kind: "request_confirmation", status: "pending", sourceRunId: input.runId, addresseeAgentId: reviewer.id, + effectiveResolverPolicy: "anyone", resolverPolicyProvenance: "inherited", + payload: { version: 1, prompt: "Review the work", target: { type: "custom", key: "native_completion_review", revisionId: decision.id } }, + }).returning(); + const wakeId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: wakeId, companyId: f.companyId, agentId: reviewer.id, source: "automation", status: "queued" }); + const seeded = await run(f, { agentId: reviewer.id, status: "queued", runtimeMode: "native", nativeIssueId: f.issueId, wakeupRequestId: wakeId, + contextSnapshot: { ...runValues().contextSnapshot, nativeReviewInteractionId: interaction.id, nativeReviewDecisionId: decision.id }, + }); + 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 claim = { run: before, agentNameKey: "reviewer", claimedAt: new Date(), claimValues: {} }; + await start(); await flush(); await quiet(); + expect(await claimQueuedNativeReviewRun(db, claim)).toMatchObject({ id: seeded.id, status: "running" }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${seeded.id}:working:${f.endpointId}`); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId).toBe(seeded.id); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0].status).toBe("claimed"); + await quiet(); + expect(await claimQueuedNativeReviewRun(db, claim)).toBeNull(); await quiet(); + }); it("wakes from native committed-result projection and leaves repeated finalization quiet", async () => { const { row } = await nativeFixture(); await start(); await flush(); await quiet(); diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts index d8e30032d73..82741d5d548 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts @@ -7,6 +7,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "../../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; import { createQueuedCommentIssueLockWriter } from "./queued-comment-postgres.js"; import type { QueuedCommentQueuePostgresAdapterDeps } from "./queued-comment-postgres.js"; import { QueuedCommentMutationError } from "../application/queued-comment-use-cases.js"; @@ -19,7 +20,8 @@ import { QueuedCommentMutationError } from "../application/queued-comment-use-ca // leaving a partial write. The decision branching itself is proven against // plain facts in `domain/policy.test.ts`; the use-case orchestration is // proven against a mocked port in `application/queued-comment-use-cases.test.ts`. -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_QUEUED_COMMENT_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -39,8 +41,8 @@ describeEmbeddedPostgres("queued-comment postgres adapter", () => { }; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-queued-comment-postgres-adapter-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-queued-comment-postgres-adapter-"); + db = createDb(external ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -56,6 +58,7 @@ describeEmbeddedPostgres("queued-comment postgres adapter", () => { }); afterAll(async () => { + await db?.$client.end(); await tempDb?.cleanup(); }); diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index 2588636e4bc..e2c0e2f740f 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { and, eq, inArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agentWakeupRequests, agents, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; @@ -45,7 +46,7 @@ export type QueuedCommentQueuePostgresAdapterDeps = { syncCommentExternalObjectsSafely(commentId: string, tx: Db): Promise; }; -function buildTransaction(tx: Db, companyId: string, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentQueueTransaction { +function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentQueueTransaction { return { async updateCommentBody({ issueId, commentId, body, updatedAt }) { const updated = await tx @@ -72,6 +73,7 @@ function buildTransaction(tx: Db, companyId: string, deps: QueuedCommentQueuePos }, async updateQueueRunCommentIds({ queueRunId, contextSnapshot, ids, updatedAt }) { + await notifyChatRunMilestones(tx, { companyId, issueId }); const row = await tx .update(heartbeatRuns) .set({ contextSnapshot: withQueuedCommentIdsInRunContext(contextSnapshot, ids), updatedAt }) @@ -82,6 +84,7 @@ function buildTransaction(tx: Db, companyId: string, deps: QueuedCommentQueuePos }, async deleteComment({ issueId, commentId }) { + await notifyChatRunMilestones(tx, { companyId, issueId }); const row = await tx .delete(issueComments) .where(and(eq(issueComments.id, commentId), eq(issueComments.issueId, issueId), eq(issueComments.companyId, companyId))) @@ -98,6 +101,7 @@ function buildTransaction(tx: Db, companyId: string, deps: QueuedCommentQueuePos }, async cancelQueueRun({ queueRunId, reason, now }) { + await notifyChatRunMilestones(tx, { companyId, issueId }); const row = await tx .update(heartbeatRuns) .set({ status: "cancelled", finishedAt: now, error: reason, errorCode: "queued_comment_discarded", updatedAt: now }) @@ -291,7 +295,7 @@ export function createQueuedCommentIssueLockWriter(db: Db, deps: QueuedCommentQu .then((rows) => rows[0] ?? null) : null; - const transaction = buildTransaction(tx, companyId, deps); + const transaction = buildTransaction(tx, companyId, input.issue.id, deps); const wake = toWakeRow(wakeRow); const queueRun = queueRunRow ? toRunRow(queueRunRow) : null; const activeRun = activeRunRow ? toRunRow(activeRunRow) : null;