fix(server): signal native milestone projection changes

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

server/src/__tests__/chat-channels.integration.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDatabaseWork } from "@paperclipai/db";
+import { restoreNativeChatReviewPresentationInTransaction } from "../services/native-runtime/native-chat-review-presentation.js";
 import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js";
 import { configureAgentLifecycle } from "../services/agent-lifecycle.js";
 import { deleteCompany } from "../services/company-deletion.js";
@@ -70400,6 +70402,40 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     };
   }
 
+  it("restores changed native review metadata once and signals only its committed change", async () => {
+    const context = await committedNativeReviewPublicationFixture("milestone-repair");
+    const events: string[] = [];
+    const unsubscribe = subscribeDatabaseWork(db, "chat-run-milestones", event => events.push(event));
+    try {
+      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, context.runId));
+      const input = { companyId: context.fixture.companyId, issueId: coordinator.issueId, runId: context.runId,
+        decisionId: coordinator.decisionId!, resultId: coordinator.resultId!, assessmentId: coordinator.assessmentId! };
+      const savedAt = new Date(Date.now() - 60_000);
+      await db.update(heartbeatRuns).set({ updatedAt: savedAt }).where(eq(heartbeatRuns.id, context.runId));
+      // Already-restored metadata still authorizes materialization by the caller,
+      // but must neither rewrite the run nor wake projection again.
+      await expect(db.transaction(tx => restoreNativeChatReviewPresentationInTransaction(tx as unknown as typeof db, input))).resolves.toBe(true);
+      expect((await readHeartbeatRunsById(context.runId))[0].updatedAt).toEqual(savedAt);
+      expect(events).toEqual([]);
+      await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` })
+        .where(eq(heartbeatRuns.id, context.runId));
+      await db.transaction(async tx => {
+        expect(await restoreNativeChatReviewPresentationInTransaction(tx as unknown as typeof db, input)).toBe(true);
+        expect(events).toEqual(["intent"]);
+      });
+      expect(events).toEqual(["intent", "settled"]);
+      expect((await readHeartbeatRunsById(context.runId))[0].resultJson).toMatchObject({
+        externalChatReviewPresentation: { decisionId: coordinator.decisionId, runId: context.runId },
+      });
+      events.length = 0;
+      await db.transaction(tx => restoreNativeChatReviewPresentationInTransaction(tx as unknown as typeof db, input));
+      expect(events).toEqual([]);
+    } finally {
+      unsubscribe();
+      await context.service.shutdown();
+    }
+  });
+
   it.each(["principal_revoked", "gate_changed"] as const)(
     "blocks a committed native review response and selected files at transport after %s",
     async (kind) => {

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, authUsers, companyMemberships, chatConversations, chatEndpoints, chatMessageLinks, chatPublications, companies, createDb,
+import { activityLog, agents, authUsers, companyMemberships, chatConversations, chatEndpoints, chatMessageLinks, chatPublications, companies, completionContracts, createDb, nativeRunFinalizations, nativeRunResults,
   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";
@@ -11,6 +11,7 @@ import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifica
 import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
 import { questionResponseDeliveryService } from "../services/question-response-delivery.js";
 import { heartbeatService } from "../services/heartbeat.js";
+import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 
 const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MILESTONE_TEST_DATABASE_URL");
@@ -66,7 +67,10 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
       await db.delete(issueThreadInteractions).where(eq(issueThreadInteractions.companyId, companyId));
       await db.delete(issueComments).where(eq(issueComments.companyId, companyId));
       await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.companyId, companyId));
+      await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId));
+      await db.delete(nativeRunResults).where(eq(nativeRunResults.companyId, companyId));
       await db.delete(heartbeatRuns).where(eq(heartbeatRuns.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));
       await db.delete(companyMemberships).where(eq(companyMemberships.companyId, companyId));
@@ -128,6 +132,41 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     ].sort());
     await quiet();
   });
+  it("wakes from native committed-result projection and leaves repeated finalization quiet", async () => {
+    const contractId = randomUUID(), resultId = randomUUID();
+    await db.insert(completionContracts).values({
+      id: contractId, companyId: f.companyId, issueId: f.issueId, revision: 1,
+      schemaVersion: "paperclip.completion-contract.v1", policyVersion: "phase6-v1", risk: "standard",
+      completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
+      contractJson: { objective: "Respond to chat" }, canonicalSha256: "milestone-contract",
+      createdByActorType: "system", createdByActorId: "test",
+    });
+    // Persisted audit-only routing can commit without a status decision. Recovery
+    // still owes the terminal run projection after the controller restarts.
+    const row = await run(f, { runtimeMode: "native", nativeIssueId: f.issueId, completionContractId: contractId,
+      resultJson: { nativeAttentionRouting: [{ decisionId: null, reasonCode: "attention_duplicate_suppressed",
+        materializedTargets: [{ kind: "existing_attention" }] }] },
+    });
+    await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId,
+      runId: row.id, completionContractId: contractId, serverFingerprint: resultId, schemaStatus: "accepted",
+      resultJson: { terminal: { runTerminalState: "cancelled" } }, canonicalSha256: "milestone-result",
+    });
+    await db.insert(nativeRunFinalizations).values({ runId: row.id, companyId: f.companyId, issueId: f.issueId,
+      phase: "committed", resultId,
+    });
+    await start(); await flush(); await quiet();
+    await finalizeNativeRun({ db, runId: row.id, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    await flush();
+    expect((await publications()).map(publication => publication.idempotencyKey).sort()).toEqual([
+      `run:${row.id}:failed:${f.endpointId}`, `run:${row.id}:working:${f.endpointId}`,
+    ].sort());
+    await quiet();
+    await finalizeNativeRun({ db, runId: row.id, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    // A no-op repair must not enqueue another scan. This also catches deferred
+    // coordinator callbacks that start after the finalizer returns.
+    await quiet();
+    expect(await publications()).toHaveLength(2);
+  });
   it("projects a continuation only after the real question delivery receipt commits", async () => {
     const source = await run(f, { status: "succeeded" });
     const interactions = issueThreadInteractionService(db);

server/src/services/native-runtime/native-chat-review-presentation.ts

@@ -1,3 +1,4 @@
+import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
 import { and, eq, inArray, ne, sql } from "drizzle-orm";
 import {
   agents,
@@ -573,20 +574,28 @@ export async function restoreNativeChatReviewPresentationInTransaction(
     assessmentId: input.assessmentId,
     externalChatReviewPresentation: { ...proof, decisionId: input.decisionId },
   };
+  const runOwner = and(
+    eq(heartbeatRuns.id, input.runId),
+    eq(heartbeatRuns.companyId, input.companyId),
+    eq(heartbeatRuns.nativeIssueId, input.issueId),
+    eq(heartbeatRuns.runtimeMode, "native"),
+  );
+  const projectedResult = sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(projection)}::jsonb`;
+  const [run] = await db.select({
+    needsUpdate: sql<boolean>`${projectedResult} is distinct from ${heartbeatRuns.resultJson}`,
+  }).from(heartbeatRuns).where(runOwner).for("update");
+  if (!run) return false;
+  // A valid existing projection still permits the caller to materialize its
+  // response. Replaying recovery must not rewrite it or signal another scan.
+  if (!run.needsUpdate) return true;
+  await notifyChatRunMilestones(db, input);
   const [updated] = await db
     .update(heartbeatRuns)
     .set({
-      resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(projection)}::jsonb`,
+      resultJson: projectedResult,
       updatedAt: new Date(),
     })
-    .where(
-      and(
-        eq(heartbeatRuns.id, input.runId),
-        eq(heartbeatRuns.companyId, input.companyId),
-        eq(heartbeatRuns.nativeIssueId, input.issueId),
-        eq(heartbeatRuns.runtimeMode, "native"),
-      ),
-    )
+    .where(runOwner)
     .returning({ id: heartbeatRuns.id });
   return Boolean(updated);
 }

server/src/services/native-runtime/native-run-finalizer.ts

@@ -1,4 +1,5 @@
 import { pendingNativeGovernance } from "./native-governance.js";
+import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
 import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.js";
 import { assertNativeCaseHandoffResult } from "./native-case-handoff.js";
@@ -376,6 +377,7 @@ async function recordRetryableFailure(input: {
           : input.failureCode;
     if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
     await notifyExecutionRunWork(tx);
+    await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     await tx
       .update(nativeRunFinalizations)
       .set({
@@ -665,6 +667,7 @@ async function projectCommittedRun(input: {
     const tx = transaction as unknown as Db;
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
     await notifyExecutionRunWork(tx);
+    await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     return tx
       .update(heartbeatRuns)
       .set({
@@ -938,6 +941,7 @@ export async function repairCommittedNativeChatResponse(
         const authorizationReason = await resolveChatRunPresentationAuthorizationReason(tx, input);
         const resolved = resolveHeartbeatRunResponse({ resultJson: monitorResult });
         if (!resolved.text) return false;
+        await notifyChatRunMilestones(tx, input);
         const comment = await issueService(db).addComment(input.issueId, resolved.text,
           { agentId: run.agentId, runId: run.id }, { authorizationReason, completionReply: true }, tx);
         await tx.update(heartbeatRuns).set({
@@ -992,6 +996,7 @@ export async function repairCommittedNativeChatResponse(
           finalAgentMessage,
         });
         if (!resolved.text || resolved.decision.commentAction !== "create") return false;
+        await notifyChatRunMilestones(tx, input);
         const comment = await issueService(db).addComment(input.issueId, resolved.text,
           { agentId: run.agentId, runId: run.id },
           { authorizationReason: "internal_agent_write", completionReply: true }, tx);
@@ -1035,6 +1040,7 @@ export async function repairCommittedNativeChatResponse(
       });
       if (!resolved.text || resolved.decision.commentAction !== "create")
         return false;
+      await notifyChatRunMilestones(tx, input);
       const comment = await issueService(db).addComment(
         input.issueId,
         resolved.text,
@@ -1597,6 +1603,7 @@ export async function finalizeNativeRun(input: {
         const tx = transaction as unknown as Db;
         if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
         await notifyExecutionRunWork(tx);
+        await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
         return projectRun(tx);
       });
       if (