From 2fd84fcf6f12451d6bd7dbbf5e96a0f51c956e61 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 07:13:19 -0500 Subject: [PATCH] fix(server): wake chat milestones from native decision receipts Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 8 ++ ...26-10-25-chat-run-milestone-sleep-audit.md | 43 +++++++++++ .../__tests__/chat-run-milestone-work.test.ts | 74 ++++++++++++++++++- .../status-decision-committer.ts | 3 + 4 files changed, 125 insertions(+), 3 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 a927db1994b..2e087b418de 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1413,3 +1413,11 @@ See the slice audit for failure logs and coverage limits. The production scan remains until status-decision/effect, admission, lineage, publication, and remaining producer coverage is complete. This is local progress, not a published migration or completion of the full stack. + + +Slice 28 now also covers new native status-decision commits and pending-effect +receipts, after their no-op exits. Two regressions failed first; all 18 milestone +worker cases, nine native chat recovery integrations, and server production types +pass after repair. Real late continuation receipts now reactivate projection. +Next is heartbeat queue/admission/recovery producer coverage. Periodic projection +and all remaining inventory/remote acceptance duties remain outstanding. 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 490a246450e..7c4d06adc01 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 @@ -288,3 +288,46 @@ publication, conversation/configuration, and monitor/semantic-receipt writers from the table above. Signals added here do not establish complete producer coverage. The full original stack, unresolved earlier gates, and master reconciliation remain required; no PR has been merged by this checkpoint. + + +## Native decision and effect-receipt checkpoint + +`commitNativeStatusDecision` registers company/issue milestone intent after the +existing committed/replayed-decision exits, before inserting or applying a new +decision and its effects. `dispatchPendingNativeStatusEffects` registers intent +only when pending/failed effects exist and their targets have passed validation, +before their delivered receipt and run effect metadata are written. Both use the +existing owning transactions; no caller wrapper or new persistence is needed. +Pure committed replay and an empty pending-effect set remain unsignalled. + +Why the effect receipt matters: `readNativeContinuationSource` requires the +applied decision, delivered same-agent continuation effect, original wake intent, +dispatched wake, and responsible-user continuity. A target run can already exist +while its missing delivered receipt prevents publication. A notification only at +run insertion cannot recover that later eligibility transition. + +Two new PostgreSQL regressions failed before these integrations: +`/tmp/sleep28-native-effects-before.log`. One observed no milestone intent in the +real decision transaction; the other committed a real effect-delivery receipt but +still had no successor notice. After the repair all 18 milestone worker cases +pass (`/tmp/sleep28-native-effects-tests.log`). The new cases exercise the real +committer and effect dispatcher, outer-commit settlement, committed replay, late +lineage admission and quiet repeated recovery. The continuation is initially +suppressed even though its run and dispatch already exist, and becomes publishable +only after the real receipt producer runs. Fixtures preserve real foreign keys. + +All nine selected native chat integration cases still pass after the committer +change (`/tmp/sleep28-native-effects-acceptance.log`); expected lock contention +remains retryable. Server production typecheck passes +(`/tmp/sleep28-native-effects-typecheck.log`). These are local checks, not full +workspace or PR acceptance, and do not include test-file typechecking. + +Next concrete audit is heartbeat queue admission, coalescing, cancellation, +claim/release and wake-receipt writers in `heartbeat/queue.ts`, then the other +recovery writers. Some run mutations still use the root DB directly; place any +new transaction at the narrow mutation owner rather than wrapping execution or +provider I/O. Avoid signalling no-op periodic inspections and lease renewals. +Remaining interaction, publication, conversation/configuration, question +claim/retry, and monitor/semantic-receipt dependencies still require coverage. +The new worker is still not production-registered, and the periodic combined +projector remains until that audit and its real producer tests are complete. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 768cdf46855..284a4c80f37 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/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, completionContracts, createDb, nativeRunFinalizations, nativeRunResults, +import { activityLog, agents, agentWakeupRequests, authUsers, companyMemberships, chatConversations, chatEndpoints, chatMessageLinks, chatPublications, companies, completionContracts, createDb, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, workAssessments, subscribeDatabaseWork, 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 { commitNativeStatusDecision, dispatchPendingNativeStatusEffects } from "../services/native-runtime/status-decision-committer.js"; import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -68,6 +69,10 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu 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(statusDecisionEffects).where(eq(statusDecisionEffects.companyId, companyId)); + await db.delete(statusDecisions).where(eq(statusDecisions.companyId, companyId)); + await db.delete(workAssessments).where(eq(workAssessments.companyId, companyId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.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)); @@ -132,7 +137,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu ].sort()); await quiet(); }); - it("wakes from native committed-result projection and leaves repeated finalization quiet", async () => { + async function nativeFixture(phase = "committed") { const contractId = randomUUID(), resultId = randomUUID(); await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId, revision: 1, @@ -152,8 +157,29 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu resultJson: { terminal: { runTerminalState: "cancelled" } }, canonicalSha256: "milestone-result", }); await db.insert(nativeRunFinalizations).values({ runId: row.id, companyId: f.companyId, issueId: f.issueId, - phase: "committed", resultId, + phase, resultId, }); + return { row, contractId, resultId }; + } + async function continuationDecisionFixture() { + const native = await nativeFixture("assessing"); + const assessmentId = randomUUID(); + await db.insert(workAssessments).values({ id: assessmentId, companyId: f.companyId, issueId: f.issueId, + runId: native.row.id, contractId: native.contractId, resultId: native.resultId, triggerKind: "run_result", + triggerActorCompanyId: f.companyId, priorIssueStatus: "in_progress", priorStatusVersion: 0, + policyVersion: "phase6-v1", assessmentJson: {}, inputDigest: assessmentId, + }); + return { companyId: f.companyId, issueId: f.issueId, runId: native.row.id, assessmentId, + priorStatus: "in_progress", priorStatusVersion: 0, priorDecisionId: null, + decision: { policyVersion: "phase6-v1", statusAction: "preserve" as const, toStatus: "in_progress" as const, + reasonCode: "live_continuation_registered", unblockDescriptor: null, + effects: [{ kind: "enqueue_continuation" as const, agentId: f.agentId, continuationKind: "same_agent" as const, + summary: "Continue responding", idempotencyKey: `continuation:${native.row.id}` }], + }, + }; + } + it("wakes from native committed-result projection and leaves repeated finalization quiet", async () => { + const { row } = await nativeFixture(); await start(); await flush(); await quiet(); await finalizeNativeRun({ db, runId: row.id, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); await flush(); @@ -167,6 +193,48 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await quiet(); expect(await publications()).toHaveLength(2); }); + it("signals a new native decision at outer commit but not an already committed replay", async () => { + const input = await continuationDecisionFixture(); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, "chat-run-milestones", event => events.push(event)); + try { + await db.transaction(async tx => { + await commitNativeStatusDecision({ db: tx as unknown as typeof db, ...input }); + expect(events).toEqual(["intent"]); + }); + expect(events).toEqual(["intent", "settled"]); + events.length = 0; + await expect(commitNativeStatusDecision({ db, ...input })).resolves.toMatchObject({ replayed: true }); + expect(events).toEqual([]); + } finally { unsubscribe(); } + }); + it("wakes a suppressed continuation when its native effect delivery receipt commits", async () => { + const input = await continuationDecisionFixture(); + const { decision } = await commitNativeStatusDecision({ db, ...input }); + const [intent] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + // Reconstruct an interrupted delivery with its already admitted successor. + // The target has no direct chat identity: only the delivered ledger can + // authorize inheriting the source run's conversation. + await db.update(statusDecisionEffects).set({ deliveryState: "pending", deliveredAt: null }) + .where(eq(statusDecisionEffects.decisionId, decision.id)); + const target = await run(f, { status: "queued", contextSnapshot: { + issueId: f.issueId, source: "native_status_decision", nativeStatusWakeIntentId: intent.id, + } }); + await db.update(agentWakeupRequests).set({ status: "coalesced", runId: target.id }).where(eq(agentWakeupRequests.id, intent.id)); + await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + reason: "issue_status_changed", requestedByActorType: "system", requestedByActorId: `native-status-wake-dispatch:${intent.id}`, + runId: target.id, status: "queued", idempotencyKey: intent.idempotencyKey, payload: { issueId: f.issueId }, + }); + await start(); await flush(); + expect((await publications()).some(row => row.idempotencyKey.startsWith(`run:${target.id}:`))).toBe(false); + await quiet(); + await dispatchPendingNativeStatusEffects({ db, ...input, decisionId: decision.id }); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${target.id}:queued:${f.endpointId}`); + await quiet(); + await expect(dispatchPendingNativeStatusEffects({ db, ...input, decisionId: decision.id })).resolves.toEqual([]); + await quiet(); + }); it("projects a continuation only after the real question delivery receipt commits", async () => { const source = await run(f, { status: "succeeded" }); const interactions = issueThreadInteractionService(db); diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index eafd64245c9..6dd4f18f13c 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,3 +1,4 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; import { hasPublishedGitHubReview } from "../chat-channels/github/review-wait.js"; import { pendingNativeGovernance } from "./native-governance.js"; @@ -125,6 +126,7 @@ export async function dispatchPendingNativeStatusEffects(input: { targetId: effect.targetId, }); } + await notifyChatRunMilestones(tx, input); await tx .update(statusDecisionEffects) .set({ @@ -1931,6 +1933,7 @@ export async function commitNativeStatusDecision(input: { await recordCoordinatorDecision(decisionRow.id); return { decision: decisionRow, issue, replayed: true }; } + await notifyChatRunMilestones(tx, input); if (!decisionRow) { const latestDecisionVersion = await tx .select({