From c4beddad1e8cd0af1784f8367ffd05523255b5b3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 00:24:07 -0500 Subject: [PATCH] fix: scope native response wakes to affected runs and tasks Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 14 +++++++ ...ative-response-presentation-sleep-audit.md | 41 +++++++++++++++++++ .../chat-channels.integration.test.ts | 32 ++++++++++++++- .../chat-run-milestone-notifications.ts | 5 ++- server/src/services/heartbeat/queue.ts | 7 ++-- .../src/services/heartbeat/run-completion.ts | 2 +- .../services/heartbeat/run-lifecycle.test.ts | 19 ++++++++- .../src/services/heartbeat/run-lifecycle.ts | 5 ++- 8 files changed, 114 insertions(+), 11 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 67938973e3a..9a3deaee3e0 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2350,3 +2350,17 @@ active. The startup suite passes 66 cases, and two database regressions confirm periodic calls defer to the new worker. Workspace types pass. Full local tests remain in progress with embedded-database initialization failures; build and remote checks remain pending. The stack goal is still incomplete. + + +## Slice 48 review repair, 2026-10-26 + +Draft #17696 is open and attached. Review identified company-wide response reloads +from routine run-status notifications; three real database regressions reproduced +it. Status/claim/release wakes now retain task or exact-run scope while task +lineage remains task-wide. All 34 heartbeat lifecycle and 63 response/authority/ +admission tests pass, as do server types and the workspace build. The full local +suite was stopped after repeated embedded PostgreSQL startup failures and remains +unproven. Remote failures at `79bf36cbf8` include an early startup-settlement test +assertion (repaired), a workspace staging timeout, and an E2E runner shutdown. +See the response audit for logs. Fresh-head CI/review, earlier stack findings, +master reconciliation, and all remaining original duties stay open. Nothing merged. diff --git a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md index 84ba37c2306..12ea5b1fb71 100644 --- a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md +++ b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md @@ -335,3 +335,44 @@ Workspace typecheck passes (`/tmp/sleep48-full-typecheck.log`). The full local s is still running and has reported embedded PostgreSQL initialization failures; its final result and build status remain pending. No full-suite or remote check acceptance is claimed at this checkpoint. + + +## Review repair and current verification, 2026-10-26 + +Published draft #17696 at `79bf36cbf8`. Greptile found a real precision gap: +`writeRunStatus` still sent a company-only milestone hint. Every unrelated run +status could reconstruct dormant response history. Three real-PostgreSQL cases +(native binding, legacy context binding, and taskless run) failed before repair +(`/tmp/sleep48-scope-before.log`). + +The milestone contract now accepts an optional run identity. Status producers +supply the task when bound and the exact run otherwise. The response fanout +prefers task scope over run scope, because lineage changes can authorize other +runs on that task. Known run claim/release and completion producers also preserve +an exact-run fallback. Existing company configuration invalidations remain broad. +No new durable rows or caller recovery wrappers are introduced. + +All 34 heartbeat lifecycle tests pass after repair. Two real worker regressions +retain a dormant eligible response, send unrelated task/run wakes, and observe +one actual candidate SQL query at the postgres.js debug boundary, without +rechecking the dormant row. The outer-commit/rollback cases additionally send a +different run identity within the same task and retain task-level recovery. +An initial test mistakenly counted Drizzle subquery builder construction as a +second SQL statement; driver-level observation corrected the measurement. + +The response/authority/admission selection now passes all 63 cases +(`/tmp/sleep48-scope-acceptance.log`). Server typechecking and the workspace build +pass (`/tmp/sleep48-scope-typecheck-v3.log`, `/tmp/sleep48-scope-build.log`). +The previous full workspace typecheck passed at `79bf36cbf8`. The full local +Vitest run was explicitly stopped after repeated embedded PostgreSQL bootstrap +failures; `/tmp/sleep48-full-tests.log` preserves those failures. It is not a +full-suite pass. Focused database acceptance uses the validated external +PostgreSQL path and isolated disposable test databases. + +Remote CI at `79bf36cbf8` also exposed a late-journal test that assumed worker +startup drains all candidates in one batch. It now waits for actual settlement, +as the shared worker deliberately processes fair batches. The workspaces-b job +hit a five-second sandbox staging test timeout; E2E shard five received a runner +shutdown signal. These failures remain recorded in `/tmp/sleep48-ci-chat3.log`, +`/tmp/sleep48-ci-workspaces-b.log`, and `/tmp/sleep48-ci-e2e5.log`. Fresh-head CI and +review are required after the repair; no failing gate is waived. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index d13c43499b5..49d8efaca8c 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -72675,6 +72675,34 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { worker.stop(); await coordinator.stop(); await replacement?.service.shutdown(); } }); + it.each(["issue", "run"])("does not reload dormant accepted-response history for an unrelated %s wake", async scope => { + const context = await committedChatResponseRecoveryFixture("telegram"); + await context.service.shutdown(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const work = nativeResponsePresentationWork(db); + const worker = work.register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 10_000 }).toBeNull(); + expect(await work.load({ companyId: context.fixture.companyId, runId: context.runId })).toHaveLength(1); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { + if (query.startsWith("select") && query.includes("native_run_finalizations")) queries.push(query); + }; + try { + await db.transaction(tx => notifyChatRunMilestones(tx, { companyId: context.fixture.companyId, + ...(scope === "issue" ? { issueId: randomUUID() } : { runId: randomUUID() }) })); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 5000 }).toEqual({ active: 0, next: null }); + expect(queries).toHaveLength(1); + } finally { db.$client.options.debug = previousDebug; } + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); } + }); + it.each(["commit", "rollback"])("keeps accepted-response work behind the outer %s boundary", async outcome => { const context = await committedChatResponseRecoveryFixture("telegram"); await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, context.runId)); @@ -72686,7 +72714,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); const transaction = db.transaction(async tx => { await tx.transaction(async nested => { - await notifyChatRunMilestones(nested, { companyId: context.fixture.companyId, issueId: context.issue.id }); + await notifyChatRunMilestones(nested, { companyId: context.fixture.companyId, issueId: context.issue.id, runId: randomUUID() }); await nested.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, context.runId)); }); expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); @@ -72886,7 +72914,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); try { await worker.ready; - expect(coordinator.nextWakeAt()).toBeNull(); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 10_000 }).toBeNull(); expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); const port = new PaperclipControlPlanePort(db, { companyId: context.binding.companyId, issueId: context.binding.issueId, diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 8b0d51d37fe..00cb9f0414a 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -6,10 +6,11 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio /** Lineage changes can authorize descendants, so invalidate the issue rather * than just the run whose row changed. Configuration may invalidate a company. */ -export type ChatRunMilestoneScope = { companyId: string; issueId?: string }; +export type ChatRunMilestoneScope = { companyId: string; issueId?: string; runId?: string }; export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { - await notifyNativeResponsePresentation(tx, scope); + await notifyNativeResponsePresentation(tx, { companyId: scope.companyId, issueId: scope.issueId, + runId: scope.issueId ? undefined : scope.runId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify({ companyId: scope.companyId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index f3d114dd222..29007f364c4 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -974,7 +974,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "queued"))) .for("update"); if (!currentRun) return null; - await notifyChatRunMilestones(tx, { companyId: run.companyId }); + await notifyChatRunMilestones(tx, { companyId: run.companyId, runId: run.id, + issueId: run.nativeIssueId ?? run.issueId ?? readNonEmptyString(parseObject(run.contextSnapshot).issueId) ?? undefined }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); @@ -1623,7 +1624,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende )).for("update"); if (!lockedRun || lockedRun.status !== "queued") return null; await withExternalAdmissionGuard(claimTx, run.companyId, run.agentId, () => assertNoExternalOverlap(claimTx, run.companyId, run.agentId)); - await notifyChatRunMilestones(claimTx, { companyId: run.companyId, issueId: issueId ?? undefined }); + await notifyChatRunMilestones(claimTx, { companyId: run.companyId, runId: run.id, issueId: issueId ?? undefined }); const claimedRun = await claimTx.update(heartbeatRuns).set(claimValues).where(and( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "queued"), )).returning().then((rows) => rows[0] ?? null); @@ -1724,7 +1725,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const [candidate] = await tx.select({ companyId: heartbeatRuns.companyId, contextSnapshot: heartbeatRuns.contextSnapshot }) .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.status, "running"))).for("update"); if (!candidate) return; - await notifyChatRunMilestones(tx, { companyId: candidate.companyId, + await notifyChatRunMilestones(tx, { companyId: candidate.companyId, runId, issueId: readNonEmptyString(parseObject(candidate.contextSnapshot).issueId) ?? undefined }); const released = await tx .update(heartbeatRuns) diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 17a4271d40e..b16d4c18ff7 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -891,7 +891,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC } await db.transaction(async tx => { - await notifyChatRunMilestones(tx, { companyId: livenessRun.companyId, issueId: issueId ?? undefined }); + await notifyChatRunMilestones(tx, { companyId: livenessRun.companyId, runId: livenessRun.id, issueId: issueId ?? undefined }); await tx.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision })}::jsonb`, updatedAt: new Date(), diff --git a/server/src/services/heartbeat/run-lifecycle.test.ts b/server/src/services/heartbeat/run-lifecycle.test.ts index 1360bafe0c3..6d862394ed2 100644 --- a/server/src/services/heartbeat/run-lifecycle.test.ts +++ b/server/src/services/heartbeat/run-lifecycle.test.ts @@ -1,7 +1,7 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { randomBytes } from "node:crypto"; import type { Db } from "@paperclipai/db"; -import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns, issues, subscribeDatabaseWork } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -241,6 +241,23 @@ describe.skipIf(!support.supported)("heartbeat lifecycle database wiring", () => } finally { unsubscribe(); } }); + it.each(["native", "legacy", "unbound"] as const)("keeps %s run-status response recovery scoped to its task or run", async binding => { + const { run } = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: run.companyId, title: "Scope owner" }).returning(); + await db.update(heartbeatRuns).set(binding === "native" ? { nativeIssueId: issue.id } + : binding === "legacy" ? { contextSnapshot: { issueId: issue.id } } : { contextSnapshot: {} }).where(eq(heartbeatRuns.id, run.id)); + const hints: unknown[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.nativeResponsePresentation, (event, scopes) => { + if (event === "settled") for (const hint of scopes ?? []) hints.push(JSON.parse(hint)); + }); + try { + await createHeartbeatLifecycle(db, callbacks(db)).setRunStatusIfRunning(run.id, "succeeded", { finishedAt: new Date() }); + expect(hints).toEqual([binding === "unbound" ? { companyId: run.companyId, runId: run.id } + : { companyId: run.companyId, issueId: issue.id }]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded"); + } finally { unsubscribe(); } + }); + it("does not request accounting for an unchanged running status", async () => { const { run } = await fixture(); const settled = vi.fn(); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 3520380c178..234adca16e5 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1016,7 +1016,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl return recordHeartbeatRunRuntimeProgress(currentRun, update, issueId); } - async function writeRunStatus(status: string, previous: Pick | null, + async function writeRunStatus(status: string, previous: Pick | null, write: (tx: Db) => Promise) { const companyId = previous?.companyId; // Register on the outer transaction before the terminal write. An unknown @@ -1030,7 +1030,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl if (previous) await notifyCancelledWorkspacePreservation(tx, { companyId: previous.companyId, runId: previous.id }); if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId); await notifyExecutionRunWork(tx); - if (companyId) await notifyChatRunMilestones(tx, { companyId }); + if (previous) await notifyChatRunMilestones(tx, { companyId: previous.companyId, runId: previous.id, + issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined }); return write(tx); }); }