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 bac57a86133..34f670ba209 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 @@ -662,3 +662,62 @@ Remote gates refreshed at this checkpoint: #17490 at `31bcd1988823` still has failed E2E shard 3 and aggregate checks; #17590 at `e501d6c99d72` still has queued Paperclip Review; #17592 at `f53ae3d6d27c` still has neutral Paperclip Review. These are not claimed ready and no merge was attempted. + +## Publication outcome and operator-resolution checkpoint + +`notifyChatPublicationWork` now accepts an optional company/issue scope. It +retains its existing delivery and receipt notifications. Callers supply that +scope when a publication outcome can change run lineage or release the run's +provider response slot. The shared scheduler still receives only a domain scope; +it does not inspect provider payloads or select presentation policy. + +The publication worker supplies the scope for successful receipts, cancellations +and terminal preparation failures. Its attempt claim, lease renewal and paused +attempt restoration do not signal milestone work. The failure finalizer supplies +it for failed/cancelled outcomes, while retry and delivery_unknown retain the +response slot. Operator mark-delivered/cancel resolution supplies the scope; +retry-anyway does not authorize presentation until its actual delivery settles. +All additions use existing transactions and execute before the relevant write. +Provider I/O remains outside the receipt transaction. No table was added. + +Real PostgreSQL regression evidence: + +- `/tmp/sleep28-publication-preparation-before.log`: initial fixture failed its + setup expectation because answering a question correctly cancels an unsent + prompt. This was not a production failure. The corrected fixture models a + claimed send still in flight when the user answers through the UI. +- `/tmp/sleep28-publication-failure-before.log`: definite delivery failure left + the eligible source failure notice asleep. Retry and unknown-outcome controls + passed and retained suppression. The new scoped signal fixes the failure. +- `/tmp/sleep28-prompt-receipt-tests.log`: a user answers during the controlled + provider send, the real question delivery saves a target receipt, and the + worker consumes that wake while the prompt is still streaming. Only the later + real publication receipt makes the target notice eligible. +- `/tmp/sleep28-prompt-receipt-without-signal.log`: removing only the new receipt + signal reproduces the missing target notice. The source was restored in + finally. Transport and credential authorization ports are controlled; actual + attempt leases, question receipts and publication commit use PostgreSQL. This + test does not establish provider authentication or live network behavior. +- `/tmp/sleep28-publication-resolution-before.log`: both cancel and + mark-delivered operator resolutions failed to wake their newly eligible notice. + The real resolution owner now signals before its state change. Cancel exposes + the source failure; marking delivered enables the question continuation. +- `/tmp/sleep28-publication-final-tests.log`: all 48 milestone and 35 error-policy + cases passed. The three arbitration cases did not run because their embedded + PostgreSQL setup failed after its own startup retries. That suite now supports + guarded `PAPERCLIP_ARBITRATION_TEST_DATABASE_URL`, closes its client, and uses + unique company prefixes. `/tmp/sleep28-arbitration-tests.log` shows all three + pass against dedicated `paperclip_sleep28_arbitration_20261025_test`. +- `/tmp/sleep28-publication-acceptance.log`: four existing chat integration cases + pass: malformed/transient preparation, parsed Discord question continuation, + expired provider-accepted receipt ownership, and ambiguous close confirmation. +- `/tmp/sleep28-publication-final-typecheck.log`: production server typecheck + passes. Test files are excluded from that configuration. Diff checks pass. + +This covers the primary publication worker/failure/resolution owners. Interaction +mutation and cancellation, provider-specific receipt writers, inbound/binding +changes and configuration producers remain to audit. In particular, a helper +called after an interaction mutation cannot replace intent before that earlier +mutation. Production registration and removal of the combined periodic owner +remain pending. Publish this slice as a draft for CI feedback, not as a completed +sleep migration or merge-ready PR. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 91d9208cd55..c71a6001052 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/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[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[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); diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index b5f6648b646..8a4c8f7ed7a 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/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({ diff --git a/server/src/services/chat-channels/publications/resolution.ts b/server/src/services/chat-channels/publications/resolution.ts index f6e50840635..b782e01d184 100644 --- a/server/src/services/chat-channels/publications/resolution.ts +++ b/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) diff --git a/server/src/services/chat-channels/publications/worker.ts b/server/src/services/chat-channels/publications/worker.ts index 630a97a8e0b..013ef9b3327 100644 --- a/server/src/services/chat-channels/publications/worker.ts +++ b/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) diff --git a/server/src/services/chat-interaction-arbitration.test.ts b/server/src/services/chat-interaction-arbitration.test.ts index 4addb3b453e..3e32fb04317 100644 --- a/server/src/services/chat-interaction-arbitration.test.ts +++ b/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>; + let temporary: Awaited> | undefined; let db: ReturnType; - 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) => ({ diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 71b00c50cf0..7a03bbcef51 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/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 { await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInbound); } -export async function notifyChatPublicationWork(tx: object): Promise { +export async function notifyChatPublicationWork( + tx: object, + milestoneScope?: { companyId: string; issueId: string | null }, +): Promise { 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 = {