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 951a5c9472d..d890002e280 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 @@ -2,7 +2,8 @@ Baseline: Teams attempt recovery #17592 at f53ae3d6d2. Branch: codex/sleep-28-chat-run-milestones. -Status: initial owner inspection; producer audit, implementation and acceptance pending. +Status: projection owners separated and lifecycle cursor regression repaired; +transaction producer coverage, worker registration and final acceptance pending. ## Separate obligations discovered @@ -67,3 +68,80 @@ company fairness, bounded work and existing publication deduplication. All earlier stack remote gates, #17490 E2E cancellation, #17580 unresolved cleanup policy, and final master reconciliation remain required. No merge authority or full-goal completion is implied by this audit. + + +## Projection boundary checkpoint + +`enqueueChatRunLifecycleMilestones` now owns only coarse lifecycle projection. +`enqueueSafeNativeChatProgress` exposes the existing native progress owner. +The existing `enqueueChatRunMilestones` entry point still combines them with +lifecycle priority, the same 24-hour default window, and one capped insertion +budget. Production still calls that combined entry point. No periodic owner has +been removed prematurely; this checkpoint is not completion of slice 28. + +The lifecycle cursor now carries PostgreSQL's timestamp text through a typed +`(updated_at, run_id, conversation_id)` tuple comparison. With 30 unauthorized +candidates at a submillisecond timestamp, the old Date cursor repeated a page +indefinitely, never reaching valid notices. The test caps candidate page reads +so reproducing the defect cannot leave an asynchronous SQL loop after a timeout. +Native progress has its own timestamp cursor; its precision audit belongs to +that next migration and is still required. + +### Producer coverage still to implement + +The current readers establish these dependencies. This table identifies signal +owners to audit; it does not claim those signals are already installed. + +| Dependency | Owning paths and required wake | +| --- | --- | +| Run admission/status/issue binding/error code | `heartbeat/queue.ts`, `heartbeat/run-lifecycle.ts`, and native/legacy recovery writers. Existing `notifyExecutionRunWork` covers only some paths. Audit raw writes before adding milestone coverage. | +| Presentation decision and native result | `heartbeat/run-completion.ts`, native finalizers, and semantic receipt writers. Succeeded alone is insufficient: presentation may settle later. Source monitor receipts can authorize a successor without changing the successor row. | +| Question response claim and terminal receipt | `question-response-delivery.ts`: claim moves to delivering; `recordTerminal` owns a transaction that records target run/status. A late receipt can redirect an existing run's publication destination. Lease timestamp renewal is not a new presentation obligation. | +| Interaction state and source-run ownership | `issue-thread-interactions.ts`, plus recovery writers. Pending agent-authored questions suppress generic finals even before publication; resolved prompts can retain that suppression. | +| Published prompts, final comments, and close controls | Publication settlement and `issues.ts`. Published prompt receipts enable lineage; authored finals suppress generic completion; a published close requires fresh independently admitted sources before later presentation. Publication notifications must avoid a global scan feedback loop. | +| Inbound comment links and conversation binding | Ingress and conversation owners. The run's source string alone grants no authority. Endpoint assignment, publication mode, conversation state/issue and verified comment links matter. | +| Continuation wake/decision/effect receipts | `native-continuation-delivery.ts` reads applied status decisions, delivered continuation effects, the original wake and dispatched wake, plus responsible-user continuity. Audit their writers; a run notification before these commits is too early. | +| Monitor wake and source semantic receipt | `native-monitor-delivery.ts` requires an admitted monitor wake plus the source run's exact saved `set_task_monitor` receipt. Audit both writers, including late completion. | +| Board reply identity and provider configuration | `slackBoardReplyBindings`, endpoint configuration, and upgraded GitHub configuration affect selection. Scope configuration invalidation to its company/endpoints; preserve each provider's existing policy. | + +`resolveChatOriginPublicationBindings` also follows arbitrarily long verified +lineage with cycle rejection. Changing an ancestor can affect a successor; a +strict run-ID-only invalidation is insufficient for all producer families. +Issue-scoped reconstruction is appropriate for lineage changes. Run-only hints +can be used only where the writer cannot affect descendants. Shared scheduling +must not decode provider context or implement this authorization graph. + +The coarse projector currently waits on its writes and does not use skip-locked +selection. A zero insertion count may be an authoritative suppression or a +concurrent idempotent insert; database failures must retain retry. Native progress +has different lock/contention behavior and must not inherit that conclusion. + +### Verification and failure record + +- Initial owner split: nine selected PostgreSQL integration cases passed, + including separate-owner/combined-budget checks and six provider/surface cases. +- First cursor reproduction incorrectly guarded every `db.select`, including + legitimate authorization reads, and ran against retained earlier fixtures. + Both precision variants failed that overly broad guard; this was not evidence + of the production cursor defect. Log: `/tmp/sleep28-precision-before.log`. +- Corrected guard counts candidate-page reads. In a fresh database the millisecond + case passed; the submillisecond case failed after 100 repeating pages. Log: + `/tmp/sleep28-precision-clean-before.log`. +- After retaining PostgreSQL precision, ten selected integration cases passed in + a fresh database. Log: `/tmp/sleep28-projection-fixed-tests.log`. +- Broader final selection: 31 integration cases passed (1,398 unselected), + covering milestone authority, native progress, lock contention, question + continuations and authored finals. Log: `/tmp/sleep28-projection-acceptance.log`. + These overlap the earlier ten and are not additive. +- All 20 publication policy unit cases passed; server production typecheck passed. + Logs: `/tmp/sleep28-publication-unit.log`, `/tmp/sleep28-server-typecheck.log`. + Server typecheck excludes test files. No full-workspace acceptance or new PR + readiness is claimed for this preparatory checkpoint. + +Parent #17592 at `f53ae3d6d27c2642e7dfe647a2c0a9bbf2d582ac` now has passing +Actions/security/Greptile gates, no inline review findings, and a neutral +Paperclip Review result. Neutral is not a completed positive review. #17590 at +`e501d6c99d721d6428f8c490206417765c76f4f6` has passing gates with Paperclip Review +queued at inspection. #17586 also has a neutral Paperclip Review. Final review +acceptance remains separate. Refreshed master is `2e47c42fad`; stack reconciliation +remains required. No stack PR was merged. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index d434d69781e..64f6da68186 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -186,6 +186,8 @@ import { } from "../services/chat-interaction-publications.js"; import { enqueueChatRunMilestones, + enqueueChatRunLifecycleMilestones, + enqueueSafeNativeChatProgress, githubRunReplyState, resolveChatRunPresentationAuthorizationReason, CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON, @@ -34438,7 +34440,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ]); }); - it("drains bounded Slack chat-origin milestones without admitting an internal issue run", async () => { + it.each(["milliseconds", "microseconds"])("drains bounded Slack chat-origin milestones without admitting an internal issue run (%s)", async (precision) => { const fixture = await seedCompany(); const { callbacks, endpoint } = await configuredSlackEndpoint(fixture); const thread = makeThread({ @@ -34533,15 +34535,32 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { })), ]); - const inserted: number[] = []; - for (let index = 0; index < 4; index += 1) { - inserted.push( - await enqueueChatRunMilestones(db, { - since: new Date("2026-09-05T14:00:00.000Z"), - limit: 1, - }), - ); + if (precision === "microseconds") { + await db.update(heartbeatRuns).set({ + updatedAt: sql`'2026-09-05T14:59:00.000123Z'::timestamptz`, + }).where(inArray(heartbeatRuns.id, resolverRejectedRuns.map(run => run.id))); } + // Fail a non-advancing pagination loop explicitly instead of leaving a + // background query chain running after the test timeout. + const select = db.select.bind(db); + let reads = 0; + const readGuard = vi.spyOn(db, "select").mockImplementation((...args) => { + if (args[0] && "runUpdatedAt" in args[0] && ++reads > 100) { + throw new Error("Milestone projection did not advance past rejected candidates"); + } + return select(...args); + }); + const inserted: number[] = []; + try { + for (let index = 0; index < 4; index += 1) { + inserted.push( + await enqueueChatRunLifecycleMilestones(db, { + since: new Date("2026-09-05T14:00:00.000Z"), + limit: 1, + }), + ); + } + } finally { readGuard.mockRestore(); } expect(inserted).toEqual([1, 1, 1, 0]); const publications = await db .select() @@ -63297,6 +63316,36 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it.each(["separate owners", "combined budget"] as const)( + "keeps lifecycle and native progress projection independent with %s", + async (mode) => { + await enqueueChatRunMilestones(db, { since: new Date(0) }); + const context = await safeNativeProgressFixture("telegram", mode === "separate owners" ? "111" : "112"); + try { + const progressRun = await context.createRun("ongoing work"); + await context.addEvent(progressRun); + const failedRun = await context.createRun("failed work"); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, failedRun.runId)); + const input = { since: new Date(0), limit: 1 }; + await expect((mode === "separate owners" ? enqueueChatRunLifecycleMilestones : enqueueChatRunMilestones)(db, input)).resolves.toBe(1); + const rows = () => db.select({ key: chatPublications.idempotencyKey }).from(chatPublications) + .where(eq(chatPublications.conversationId, context.conversation.id)); + expect((await rows()).map(row => row.key)).toContain(`run:${failedRun.runId}:failed:${context.endpoint.id}`); + expect((await rows()).some(row => row.key.includes(":native:"))).toBe(false); + if (mode === "separate owners") { + await expect(enqueueChatRunLifecycleMilestones(db, input)).resolves.toBe(0); + } + await expect((mode === "separate owners" ? enqueueSafeNativeChatProgress : enqueueChatRunMilestones)(db, input)).resolves.toBe(1); + expect((await rows()).filter(row => row.key.includes(":native:"))).toEqual([ + { key: `run:${progressRun.runId}:working:${context.endpoint.id}:native:making_progress:1` }, + ]); + await expect(enqueueChatRunMilestones(db, input)).resolves.toBe(0); + } finally { + await retirePublicationFixture(context.service, context.endpoint.id); + } + }, + ); + it.each(["issue", "run"] as const)( "skips a locked native progress %s while unrelated final and question publications deliver", async (lockTarget) => { diff --git a/server/src/services/chat-run-publications.ts b/server/src/services/chat-run-publications.ts index 74eb2ca3901..b06d191d9e2 100644 --- a/server/src/services/chat-run-publications.ts +++ b/server/src/services/chat-run-publications.ts @@ -201,7 +201,7 @@ type ChatRunMilestoneCandidate = { runId: string; runStatus: string; runErrorCode: string | null; - runUpdatedAt: Date; + runUpdatedAt: string; issueId: string; companyId: string; endpointId: string; @@ -280,10 +280,11 @@ export function safeMilestoneText(input: { * lane. The selector intentionally reads only event identity, type, sequence, * and time; native messages and payloads stay inside Paperclip. */ -async function enqueueSafeNativeChatProgress( +export async function enqueueSafeNativeChatProgress( db: Db, - input: { since: Date; limit: number }, + options: { since?: Date; limit?: number } = {}, ): Promise { + const input = chatRunProjectionWindow(options); const basePublication = alias( chatPublications, "safe_native_progress_base_publication", @@ -658,16 +659,11 @@ async function enqueueSafeNativeChatProgress( * output, errors, tool events, and reasoning stay in Paperclip. Idempotency is * keyed by run, milestone, and endpoint so polling and restarts are harmless. */ -export async function enqueueChatRunMilestones( +export async function enqueueChatRunLifecycleMilestones( db: Db, - input: { - publicBaseUrl?: string | null; - since?: Date; - limit?: number; - } = {}, + input: ChatRunProjectionInput = {}, ): Promise { - const since = input.since ?? new Date(Date.now() - 24 * 60 * 60_000); - const limit = Math.max(1, Math.min(input.limit ?? 200, 1_000)); + const { since, limit } = chatRunProjectionWindow(input); const issueIdFromContext = boundRunIssueIdSql(); const explicitlyAuthoredCommentReason = sql`( ${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol' @@ -763,7 +759,7 @@ export async function enqueueChatRunMilestones( )`; let inserted = 0; let cursor: { - updatedAt: Date; + updatedAt: string; runId: string; conversationId: string; } | null = null; @@ -780,7 +776,9 @@ export async function enqueueChatRunMilestones( runId: heartbeatRuns.id, runStatus: heartbeatRuns.status, runErrorCode: heartbeatRuns.errorCode, - runUpdatedAt: heartbeatRuns.updatedAt, + // Keep PostgreSQL precision: a Date cursor can repeatedly select a + // rejected page whose updated_at has submillisecond digits. + runUpdatedAt: sql`${heartbeatRuns.updatedAt}::text`, issueId: chatConversations.issueId, companyId: chatConversations.companyId, endpointId: chatConversations.endpointId, @@ -939,19 +937,8 @@ export async function enqueueChatRunMilestones( gte(heartbeatRuns.updatedAt, since), isNull(chatPublications.id), pageCursor - ? or( - gt(heartbeatRuns.updatedAt, pageCursor.updatedAt), - and( - eq(heartbeatRuns.updatedAt, pageCursor.updatedAt), - or( - gt(heartbeatRuns.id, pageCursor.runId), - and( - eq(heartbeatRuns.id, pageCursor.runId), - gt(chatConversations.id, pageCursor.conversationId), - ), - ), - ), - ) + ? sql`(${heartbeatRuns.updatedAt}, ${heartbeatRuns.id}, ${chatConversations.id}) > + (${pageCursor.updatedAt}::timestamptz, ${pageCursor.runId}::uuid, ${pageCursor.conversationId}::uuid)` : undefined, ), ) @@ -1064,11 +1051,30 @@ export async function enqueueChatRunMilestones( } if (rows.length < pageSize) break; } - if (inserted < limit) { - inserted += await enqueueSafeNativeChatProgress(db, { - since, - limit: limit - inserted, - }); - } return inserted; } + +type ChatRunProjectionInput = { + publicBaseUrl?: string | null; + since?: Date; + limit?: number; +}; + +function chatRunProjectionWindow(input: ChatRunProjectionInput) { + return { + since: input.since ?? new Date(Date.now() - 24 * 60 * 60_000), + limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)), + }; +} + +/** Explicit combined reconciliation retains lifecycle priority and one shared + * publication budget. Background owners can reconcile the two duties separately. */ +export async function enqueueChatRunMilestones( + db: Db, + input: ChatRunProjectionInput = {}, +): Promise { + const { since, limit } = chatRunProjectionWindow(input); + const inserted = await enqueueChatRunLifecycleMilestones(db, { ...input, since, limit }); + if (inserted >= limit) return inserted; + return inserted + await enqueueSafeNativeChatProgress(db, { since, limit: limit - inserted }); +}