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 e5e61c4f7e2..a927db1994b 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1402,3 +1402,14 @@ children pass 30 report/startup tests after propagation. Fresh remote gates stil apply. The next new reason is chat run-milestone projection, separate from wake notices, terminal confirmations, interaction continuations and GitHub recovery. The entire original inventory and older unresolved findings remain open. + + +Slice 28 native finalization checkpoint: existing native finalizer transactions +now signal milestone projection; governed review metadata restoration avoids +unchanged writes and signals only actual changes. Both missing wake and repeated +rewrite were reproduced before repair. All 16 milestone worker cases, nine +selected native chat recovery integrations, and server production types pass. +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. 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 50f48d52a1e..490a246450e 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 @@ -241,3 +241,50 @@ This checkpoint is progress on slice 28, not a new independently complete reason or a claim that milestone polling is retired. The original full PR-stack scope, older unresolved CI/review findings, final master reconciliation and each PR's latest-head acceptance remain required. + + +## Native finalization producer checkpoint + +Native finalization now registers issue-scoped milestone intent in its existing +transactions before failure recording, committed terminal projection, and the +normal final run projection. All three committed-response repair branches signal +only after their authorization/no-op guards and immediately before comment and +presentation writes. The status arbiter, provider interpretation, and execution +policy remain with their native owners; the milestone signal carries only company +and issue identity. + +Governed review-presentation restoration also signals an actual change. Previously +it rewrote identical metadata (including updated_at) on every accepted repair. +The owner now locks the run, compares the exact proposed JSON merge against the +current result, and returns successfully without a write or signal when unchanged. +It still returns true for an already valid projection: callers may need to +materialize the response even when its metadata is already present. Existing +coordinator/proof/digest/company checks remain before this decision. + +Evidence at this checkpoint: + +- The new real-finalizer regression failed before the signal: a persisted committed + native cancellation projected its terminal run state but never generated the + terminal chat notice. Log: `/tmp/sleep28-native-before.log`. +- After wiring, all 16 milestone-worker PostgreSQL cases pass. The new case calls + the real finalizer without a manual milestone notification and verifies that + replaying the settled finalization causes no deferred worker SQL. Its input is + a seeded committed audit-only result; this is a recovery-producer proof, not a + test of accepting provider results. Log: `/tmp/sleep28-native-worker.log`. +- The new governed-review repair test failed first because an unchanged repair + advanced updated_at. Log: `/tmp/sleep28-review-before.log`. +- After repair, nine selected real PostgreSQL chat integration cases pass. They + cover unchanged metadata, changed metadata intent before the write and settlement + after commit, repeat repair, Slack/Telegram accepted-response recovery, canonical + response recovery, already-selected/deleted answer protection, contention, and + review authority changes. Expected contention is retained as retryable. + Log: `/tmp/sleep28-native-recovery-tests.log` (1,421 cases not selected). +- Server production typecheck passes: `/tmp/sleep28-native-typecheck.log`. + It excludes test files. No full-workspace or remote PR acceptance is claimed. + +The periodic projector remains enabled. Next producer audit: status-decision +commit/replay and pending-effect delivery, then admission/recovery, interaction, +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. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 64f6da68186..3f7850bfe6f 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/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) => { diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index cab52cab8a4..768cdf46855 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, 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); diff --git a/server/src/services/native-runtime/native-chat-review-presentation.ts b/server/src/services/native-runtime/native-chat-review-presentation.ts index 74cdd87c4fb..366e564acb6 100644 --- a/server/src/services/native-runtime/native-chat-review-presentation.ts +++ b/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`${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); } diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index ff5d3267a9d..ec8deb67a85 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/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 (