fix(server): wake chat milestones when publication outcomes settle

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

server/src/__tests__/chat-run-milestone-work.test.ts

@@ -21,6 +21,10 @@ import { CHAT_CONTROL_RECOVERY_ADMISSION_KEY, CHAT_CONTROL_RECOVERY_UNRESOLVED_C
 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 { createChatPublicationResolution } from "../services/chat-channels/publications/resolution.js";
+import { createChatPublicationWorker } from "../services/chat-channels/publications/worker.js";
+import { CAPABILITIES } from "../services/chat-channels/provider-catalog.js";
+import { createChatPublicationFailures } from "../services/chat-channels/publications/failures.js";
 import { PaperclipRunnerToolAuthority } from "../services/native-runtime/paperclip-runner-tool-authority.js";
 import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js";
 import { idleWorkSnapshot } from "../services/task-admission.js";
@@ -628,6 +632,127 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${targetId}:queued:${f.endpointId}`);
     await quiet();
   });
+  it("wakes a question continuation after its in-flight prompt receipt commits", async () => {
+    await db.update(chatEndpoints).set({ provider: "discord" }).where(eq(chatEndpoints.id, f.endpointId));
+    const source = await run(f, { status: "failed", contextSnapshot: { ...runValues().contextSnapshot, source: "chat:discord" } });
+    const interactions = issueThreadInteractionService(db);
+    const interaction = await interactions.create({ id: f.issueId, companyId: f.companyId }, {
+      kind: "ask_user_questions", continuationPolicy: "wake_assignee", sourceRunId: source.id,
+      title: "Choose", payload: { version: 1, title: "Choose", questions: [{
+        id: "choice", prompt: "Continue?", selectionMode: "single", required: true,
+        options: [{ id: "yes", label: "Yes" }],
+      }] },
+    }, { agentId: f.agentId, runId: source.id });
+    const prompt = (await publications()).find(row => row.idempotencyKey === `interaction:${interaction.id}:${f.endpointId}`)!;
+    const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId));
+    const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.id, f.conversationId));
+    const targetId = randomUUID();
+    let beforeReceiptKeys: string[] | undefined;
+    await start(); await flush(); await quiet();
+    const deliver = questionResponseDeliveryService(db, { heartbeat: {
+      wakeup: async () => {
+        const [target] = await db.insert(heartbeatRuns).values(runValues(f, {
+          id: targetId, status: "queued", contextSnapshot: { issueId: f.issueId },
+        })).returning();
+        return target;
+      },
+    } });
+    const post = vi.fn(async () => {
+      await interactions.answerQuestions({ id: f.issueId, companyId: f.companyId, status: "in_progress" }, interaction.id,
+        { answers: [{ questionId: "choice", optionIds: ["yes"] }] }, { userId: f.userId });
+      await deliver.deliver(interaction.id);
+      await flush();
+      beforeReceiptKeys = (await publications()).map(row => row.idempotencyKey);
+      return { id: "published-question" };
+    });
+    // Controlled transport/authorization ports; actual attempt lease, receipt
+    // transaction, question delivery receipt and milestone worker use PostgreSQL.
+    const worker = createChatPublicationWorker(db, {
+      CAPABILITIES, settleGitHubAutomaticPublication: async () => false,
+      ensureDurablePublicationTransport: async publication => publication,
+      tryProcessTeamsFilePublication: async () => false,
+      endpointRecord: async () => ({ endpoint, assignedAgentName: "Maya", connectionName: "Chat", applicationId: "test", credentialSecretRefs: [] }),
+      runtimeContextForRecord: () => ({ generation: 0, localEpoch: 0, credentialFingerprint: "test", version: "test" }),
+      hasCommittedTaskControlCompletion: async () => false, nonDirectDestinationAllowed: () => true,
+      runOwnershipMilestoneSupersessionReason: async () => null,
+      progressSupersededByFastResponse: async () => false, runProgressSupersededByPublishedInteraction: async () => false,
+      currentTaskControlPayload: async publication => publication.payload,
+      withCredentialMutationLease: async (_endpoint, operation) => operation({ assertOwned: async () => {} }),
+      closeProgressToReplace: async () => null, interactionResolutionPublicationToReplace: async () => null,
+      interactionPromptPublicationToReplace: async () => null, runPublicationToReplace: async () => null,
+      inboundWakePublicationToReplace: async () => null, taskStatusPublicationToReplace: async () => null,
+      claimPublicationTransportAuthorization: async () => ({ endpoint, conversation, authorizationActionId: null, preparedText: null }),
+      postSafePublication: post, commitTaskControlCompletion: async () => {}, receiptReactionCompletionRunId: async () => null,
+      finalizePublicationFailure: async (_publication, error) => { errors.push(error); return false; },
+      redactError: String,
+    } as Parameters<typeof createChatPublicationWorker>[1]);
+    await worker.processSelectedPublication(prompt);
+    expect(post).toHaveBeenCalledTimes(1);
+    expect((await db.select().from(chatPublications).where(eq(chatPublications.id, prompt.id)))[0]).toMatchObject({ state: "published", providerMessageId: "published-question" });
+    expect(beforeReceiptKeys).toBeDefined();
+    expect(beforeReceiptKeys).not.toContain(`run:${targetId}:queued:${f.endpointId}`);
+    await flush();
+    expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${targetId}:queued:${f.endpointId}`);
+    await quiet();
+  });
+  it.each([
+    ["retry", null], ["failed", null], ["delivery_unknown", "cancel"], ["delivery_unknown", "mark_delivered"],
+  ] as const)("rechecks resolved prompt ownership after delivery %s (%s)", async (state, resolution) => {
+    const source = await run(f, { status: "failed", errorCode: "test_run_failed" });
+    const interactions = issueThreadInteractionService(db);
+    const interaction = await interactions.create({ id: f.issueId, companyId: f.companyId }, {
+      kind: "ask_user_questions", continuationPolicy: "wake_assignee", sourceRunId: source.id,
+      title: "Choose", payload: { version: 1, title: "Choose", questions: [{
+        id: "choice", prompt: "Continue?", selectionMode: "single", required: true,
+        options: [{ id: "yes", label: "Yes" }],
+      }] },
+    }, { agentId: f.agentId, runId: source.id });
+    // Model a send already claimed when the user answers through the UI.
+    const [prompt] = await db.update(chatPublications).set({ state: "streaming", attempts: 1 })
+      .where(eq(chatPublications.idempotencyKey, `interaction:${interaction.id}:${f.endpointId}`)).returning();
+    await interactions.answerQuestions({ id: f.issueId, companyId: f.companyId, status: "in_progress" }, interaction.id,
+      { answers: [{ questionId: "choice", optionIds: ["yes"] }] }, { userId: f.userId });
+    expect(prompt).toBeDefined();
+    await start(); await flush();
+    expect((await publications()).some(row => row.idempotencyKey.startsWith(`run:${source.id}:`))).toBe(false);
+    await quiet();
+    const failures = createChatPublicationFailures(db, {
+      FailedChatRetryPublicationReadError: class extends Error {},
+      MAX_ERROR_TEXT: 1000, getTaskBaseUrl: () => null,
+      invalidateRuntime: async () => false, runtimeCallbackEndpoint: async () => null,
+    });
+    const failure = Object.assign(new Error("provider rejected the send"), { status: state === "retry" ? 429 : 400 });
+    // The finalizer receives the selection from before the attempt increment.
+    await expect(failures.finalizePublicationFailure({ ...prompt, attempts: 0 }, failure, state === "delivery_unknown")).resolves.toBe(true);
+    expect((await db.select().from(chatPublications).where(eq(chatPublications.id, prompt.id)))[0].state).toBe(state);
+    await flush();
+    expect((await publications()).some(row => row.idempotencyKey === `run:${source.id}:failed:${f.endpointId}`)).toBe(state === "failed");
+    await quiet();
+    if (resolution) {
+      const targetId = randomUUID();
+      await questionResponseDeliveryService(db, { heartbeat: { wakeup: async () => {
+        const [target] = await db.insert(heartbeatRuns).values(runValues(f, {
+          id: targetId, status: "queued", contextSnapshot: { issueId: f.issueId },
+        })).returning();
+        return target;
+      } } }).deliver(interaction.id);
+      await flush();
+      expect((await publications()).some(row => row.idempotencyKey.startsWith(`run:${targetId}:`))).toBe(false);
+      await quiet();
+      const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.endpointId));
+      const resolver = createChatPublicationResolution(db, {
+        endpointRecord: async () => ({ endpoint, assignedAgentName: "Maya", connectionName: "Chat", applicationId: "test", credentialSecretRefs: [] }),
+        withCredentialMutationLease: async (_endpoint, operation) => operation({ assertOwned: async () => {} }),
+        commitTaskControlCompletion: async () => {}, processPendingPublications: async () => {},
+      } as Parameters<typeof createChatPublicationResolution>[1]);
+      await resolver.resolvePublication(f.endpointId, prompt.id, resolution, f.userId);
+      await flush();
+      const keys = (await publications()).map(row => row.idempotencyKey);
+      expect(keys.includes(`run:${source.id}:failed:${f.endpointId}`)).toBe(resolution === "cancel");
+      expect(keys.includes(`run:${targetId}:queued:${f.endpointId}`)).toBe(resolution === "mark_delivered");
+      await quiet();
+    }
+  });
   it.each(["retry", "stale claim", "live claim", "rolled-back reset"] as const)("rechecks question prompt binding after %s", async (mode) => {
     const source = await run();
     const interactions = issueThreadInteractionService(db);

server/src/services/chat-channels/publications/failures.ts

@@ -120,7 +120,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur
             : disposition.kind === "resource_unavailable"
               ? "cancelled"
               : "failed";
-      await notifyChatPublicationWork(tx);
+      await notifyChatPublicationWork(tx, state === "failed" || state === "cancelled" ? publication : undefined);
       const [ownedPublication] = await tx
         .update(chatPublications)
         .set({

server/src/services/chat-channels/publications/resolution.ts

@@ -218,7 +218,7 @@ export function createChatPublicationResolution(db: Db, options: ResolutionOptio
             action === "retry_anyway"
           ) { await recordPhotonPublicationRetry(tx, endpointId, publication, userId); }
           const now = new Date();
-          await notifyChatPublicationWork(tx);
+          await notifyChatPublicationWork(tx, action === "retry_anyway" ? undefined : publication);
           await notifyVoicePublication(tx, initialRecord.endpoint);
           await tx
             .update(chatPublications)

server/src/services/chat-channels/publications/worker.ts

@@ -154,7 +154,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
         ? classified.retryAfterMs
         : Math.min(60_000, 2 ** Math.max(0, attempts) * 1_000);
     const result = await db.transaction(async (tx) => {
-      await notifyChatPublicationWork(tx);
+      await notifyChatPublicationWork(tx, terminal ? publication : undefined);
       const [settled] = await tx
         .update(chatPublications)
         .set({
@@ -380,7 +380,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
     }
     if (explicitSlackDelivery === "delivered") {
       await db.transaction(async (tx) => {
-        await notifyChatPublicationWork(tx);
+        await notifyChatPublicationWork(tx, publication);
         return tx.update(chatPublications).set({ state: "cancelled", redactedError: "Identical response already delivered by Slack tool", updatedAt: new Date() }).where(and(eq(chatPublications.id, publication.id), inArray(chatPublications.state, ["pending", "retry"])))
 ;
       });
@@ -507,7 +507,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
         ) {
           await publicationLease.assertOwned();
           await db.transaction(async (tx) => {
-            await notifyChatPublicationWork(tx);
+            await notifyChatPublicationWork(tx, publication);
             return tx
               .update(chatPublications)
               .set({
@@ -531,7 +531,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
           if (!endpoint.allowDirectMessages) {
             await publicationLease.assertOwned();
             await db.transaction(async (tx) => {
-              await notifyChatPublicationWork(tx);
+              await notifyChatPublicationWork(tx, publication);
               return tx
                 .update(chatPublications)
                 .set({
@@ -566,7 +566,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
           if (!nonDirectDestinationAllowed(endpoint, resource)) {
             await publicationLease.assertOwned();
             await db.transaction(async (tx) => {
-              await notifyChatPublicationWork(tx);
+              await notifyChatPublicationWork(tx, publication);
               return tx
                 .update(chatPublications)
                 .set({
@@ -605,7 +605,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
           if (!currentInteraction || currentInteraction.status !== "pending") {
             await publicationLease.assertOwned();
             await db.transaction(async (tx) => {
-              await notifyChatPublicationWork(tx);
+              await notifyChatPublicationWork(tx, publication);
               return tx
                 .update(chatPublications)
                 .set({
@@ -638,7 +638,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
         if (progressSupersessionReason) {
           await publicationLease.assertOwned();
           await db.transaction(async (tx) => {
-            await notifyChatPublicationWork(tx);
+            await notifyChatPublicationWork(tx, publication);
             return tx
               .update(chatPublications)
               .set({
@@ -710,7 +710,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
               if (!authorizationClaim) {
                 await db.transaction(async (tx) => {
                   await credentialLease.assertOwned(tx);
-                  await notifyChatPublicationWork(tx);
+                  await notifyChatPublicationWork(tx, publication);
                   await tx
                     .update(chatPublications)
                     .set({
@@ -858,7 +858,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
               await db.transaction(async (tx) => {
                 await credentialLease.assertOwned(tx);
                 const committedAt = new Date();
-                await notifyChatPublicationWork(tx);
+                await notifyChatPublicationWork(tx, publication);
                 await notifyVoicePublication(tx, authorizationClaim.endpoint);
                 const [completedPublication] = await tx
                   .update(chatPublications)

server/src/services/chat-interaction-arbitration.test.ts

@@ -15,19 +15,20 @@ import {
   toolConnections,
 } from "@paperclipai/db";
 import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
 import { hasChatRunOwnedProviderInteraction } from "./chat-interaction-arbitration.js";
 
+const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ARBITRATION_TEST_DATABASE_URL");
+
 describe("provider-owned interaction arbitration", () => {
-  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   let db: ReturnType<typeof createDb>;
-  let companyCount = 0;
   beforeAll(async () => {
-    temporary = await startEmbeddedPostgresTestDatabase(
-      "chat-interaction-owner-",
-    );
-    db = createDb(temporary.connectionString);
+    if (!externalDatabaseUrl) temporary = await startEmbeddedPostgresTestDatabase("chat-interaction-owner-");
+    db = createDb(externalDatabaseUrl ?? temporary!.connectionString);
   }, 30_000);
   afterAll(async () => {
+    await db?.$client.end();
     await temporary?.cleanup();
   });
 
@@ -43,7 +44,7 @@ describe("provider-owned interaction arbitration", () => {
       .values({
         id: companyId,
         name: "Arbitration",
-        issuePrefix: `CI${String.fromCharCode(65 + companyCount++)}`,
+        issuePrefix: `C${companyId.slice(0, 7)}`,
       });
     await db.insert(agents).values(
       [agentId, otherAgentId].map((id) => ({

server/src/services/chat-work-notifications.ts

@@ -1,3 +1,4 @@
+import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
 import { notifyChatConnectionConfiguration } from "./chat-channels/configuration-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
@@ -5,10 +6,19 @@ export async function notifyChatDeliveryWork(tx: object): Promise<void> {
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInbound);
 }
 
-export async function notifyChatPublicationWork(tx: object): Promise<void> {
+export async function notifyChatPublicationWork(
+  tx: object,
+  milestoneScope?: { companyId: string; issueId: string | null },
+): Promise<void> {
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatPublications);
   // Receipt recovery waits for the publication owner to release its claim.
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceipts);
+  // Published prompts and terminal delivery outcomes can change run lineage or
+  // release its provider response slot. Callers supply the affected issue only
+  // for those changes; transport claim/lease bookkeeping has no projection debt.
+  if (milestoneScope?.issueId) {
+    await notifyChatRunMilestones(tx, { companyId: milestoneScope.companyId, issueId: milestoneScope.issueId });
+  }
 }
 
 export const CHAT_ACTION_QUEUES = {