fix(server): signal chat milestones when queued comment lineage changes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.

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.

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();

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();
   });
 

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<void>;
 };
 
-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;