From bcf2f38ff5e25e27b5a992114c320782f41cfe9d Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 12:19:41 -0500 Subject: [PATCH] fix(server): wake native chat progress from committed work Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 17 + ...-10-25-native-chat-progress-sleep-audit.md | 236 ++++++------ .../app-private-hostname-gate.test.ts | 116 +----- .../chat-channels.integration.test.ts | 91 +++++ .../chat-native-progress-work.test.ts | 350 ++++++++++++++++++ .../voice-session-store.integration.test.ts | 2 +- server/src/app.ts | 26 +- .../chat-channels/publications/resolution.ts | 4 +- .../chat-native-progress-notifications.ts | 6 + .../src/services/chat-native-progress-work.ts | 34 ++ server/src/services/chat-native-progress.ts | 78 ++-- .../chat-run-milestone-notifications.ts | 8 +- .../services/chat-run-publication-policy.ts | 2 + server/src/services/chat-run-publications.ts | 4 + .../services/delivery-work-notifications.ts | 1 + server/src/services/heartbeat-run-events.ts | 12 +- .../native-run-coordinator-store.ts | 5 + 17 files changed, 711 insertions(+), 281 deletions(-) create mode 100644 server/src/__tests__/chat-native-progress-work.test.ts create mode 100644 server/src/services/chat-native-progress-notifications.ts create mode 100644 server/src/services/chat-native-progress-work.ts 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 de605840501..aca62227c10 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1719,3 +1719,20 @@ test so downstream Slack publication tests see their original identity. Local 40 stale-lock, 122 voice, 38 auth/session and 28 combined chat cases pass; server typecheck/build pass. See the milestone audit for failed-run dispositions. Latest-head remote gates and parent failures remain open. + +## Slice 29: native chat progress implementation + +Branch `codex/sleep-29-native-chat-progress`, parent #17606 at `ed76256b97`. +Native event producers and existing eligibility notifications now wake a scoped +worker. Progress leaves the chat timer; retained history becomes quiet. The +worker retains only unfinished batches, concrete lock contention and failures. +See [the owner audit](2026-10-25-native-chat-progress-sleep-audit.md) for producer +coverage, cursor precision repair, authorization boundaries and retained failures. + +Forty-four focused native-progress cases pass, as do 13 app coordinator cases and +the server build/typecheck. All 16 browser chat cases pass before the final +base-publication hook. The hook fixes a reproduced commit-order wake gap; its +19 worker and 13 coordinator cases pass. Final browser acceptance and +publication/checks are pending. +Parent #17606 now has passing checks and Greptile 5/5 on `ed76256b97`; all three +review threads are resolved. This is not completion of the broader stack. diff --git a/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md b/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md index 2750c81a053..5b2aeffdf18 100644 --- a/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md +++ b/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md @@ -2,141 +2,145 @@ Parent: #17606 at `ed76256b979541b962fa784016a550c32920d670`. Branch: `codex/sleep-29-native-chat-progress`. -Status: groundwork verified; production still polls. Do not mark migrated. +Status: implementation and focused PostgreSQL acceptance complete; remote acceptance pending; browser acceptance passed before the final base-publication hook. This is not a deployed sleep claim. ## Duty and owner -Native progress is the separate `projectNativeProgress` lane in the one-second -chat reconciliation coordinator. It projects a closed allowlist of durable run -event types into coarse provider-safe phrases. It never copies event messages, -payloads, tool names, arguments, or results into external publications. +Native progress used the one-second chat reconciliation cadence. It projects a +closed allowlist of durable run event types into coarse provider-safe phrases. +It never copies event messages, payloads, tool names, arguments, or results into +external publications. -`chat-native-progress.ts` owns selection, authority checks and insertion. +`chat-native-progress.ts` owns selection, authorization and insertion. `safe-native-chat-progress.ts` owns the normalized allowlist and phrase policy. -`chat-run-publication-policy.ts` holds only the existing projection window and -shared automatic-publication eligibility. The latter preserves the distinction -between upgraded GitHub bots (tool-owned replies) and legacy automatic replies. -The original export and explicit combined lifecycle/progress API stay compatible. -No generic scheduler should interpret provider payloads or decide disclosure. +`chat-run-publication-policy.ts` holds the existing projection window and shared +automatic-publication eligibility. It preserves upgraded GitHub bots' tool-owned +replies and legacy automatic replies. The original export and explicit combined +lifecycle/progress API stay compatible, including their shared insertion budget. +Startup registers the worker; it no longer invokes progress on the chat timer. +Other chat maintenance duties remain for their own migration slices. -## Existing durable contract +## Durable contract and triggers -- Native running run, matching company/agent and bound issue. -- Active/waiting conversation with an automatic endpoint assigned to that agent. -- Existing base working publication in pending/retry/streaming/published state. -- Latest allowlisted event for that run, inside the existing 24-hour window. -- Event time at least 20 seconds after the base working publication. -- Exact current origin/lineage authorization; current provider interaction and - explicitly authored final output may suppress this optional progress. -- At most one publication per phase. Twenty-second spacing compares the saved - event time with saved publication times. Passage of wall-clock time does not - make the same suppressed old event eligible. Do not invent a timer to retry it - after twenty seconds or change the cadence policy as part of this migration. -- Existing publications provide durable deduplication and outbound delivery. - Do not introduce another queue table. +The candidate requires a native running run, matching company/agent and bound +issue, an active/waiting conversation, and an automatic endpoint assigned to the +agent. Its existing base working publication must be pending, retry, streaming, +or published. The latest allowlisted event must be within the existing 24-hour +window and at least 20 seconds after the base publication. Existing lineage, +interaction arbitration, final-output suppression and delivery authorization stay +with their current owners. The ordinary direct-origin path and stricter +post-close lineage path retain their existing different authorization stages. -## Verified pagination repair +There is at most one publication per phase. Spacing compares saved event time +with saved publication times. Waiting twenty seconds does not make an old event +eligible. No timer retries cadence-suppressed events. Existing events and +publications are durable truth; there is no new queue table or schema migration. -The selector previously round-tripped event timestamps through JavaScript Date. -PostgreSQL microseconds were lost. With a full page of rejected candidates at -one fractional timestamp, every next page selected those same candidates; an -eligible event later in the ordered stream was never reached. +The two event writers register awaited intent before insertion, after duplicate +and ownership checks, inside their existing transaction boundaries: -The new regression creates 30 origin-rejected candidates with a microsecond -timestamp and a valid event after them. A query-count guard fails the broken -pagination without leaving an infinite SQL chain after the test timeout. -The fix keeps the timestamp as PostgreSQL text and uses the full timestamp/run/ -conversation tuple for continuation, preserving ordering and eligibility. +- `appendHeartbeatRunEvent` covers native ingress through + `PaperclipControlPlanePort`, direct allowlisted events and recovered trace diffs. + It reads only runtime mode and bound issue alongside the existing binding + fields. Ordinary non-progress direct log events retain their atomic-statement + path. Direct allowlisted event writes now use a transaction even for legacy + runs; only native bound runs signal progress. +- `NativeRunCoordinatorStore.appendEvent` has its own insertion path. It signals + after binding, sequence and duplicate validation, before persistence. +- `persistReprojectedWorkspaceDiffs` delegates to the first writer and is tested + through its actual recovered-trace persistence API, including duplicate replay. -- `/tmp/sleep29-progress-cursor-before.log`: fails with non-advancing pagination. -- `/tmp/sleep29-progress-cursor-acceptance.log`: 17 selected cases pass after repair. -- `/tmp/sleep29-progress-owner-acceptance.log`: 21 native-progress and combined - projection cases pass after extracting the owner. -- `/tmp/sleep29-progress-owner-typecheck.log`: server production typecheck passes. - This excludes test files. No worker-quiet or production migration claim yet. -- Database: isolated external PostgreSQL - `paperclip_sleep28_review_chat_20261025_test`, through - `PAPERCLIP_TEST_DATABASE_URL`. Tests retire their publication endpoints. +Eligibility mutations share slice 28's issue/company notification boundary. +`notifyChatRunMilestones` now invalidates both lifecycle and native progress, +without adding recovery wrappers to every caller. Both topics share the outer +transaction's identity and commit/rollback outcome. Existing writers cover +lineage receipts, run state, interactions, endpoint/resource configuration, +identity and membership. Failed-publication replay and operator retry now supply +the affected issue to `notifyChatPublicationWork`: they can restore an existing +base working publication without any new run event. The lifecycle projector also signals native progress before creating a first +working publication. A long transaction can give that publication an older +transaction-start timestamp but commit after an event was already inspected. +Routine transport claim bookkeeping retains its narrower notification behavior. -## Producer audit and remaining implementation +## Scheduling, contention and lifetime -Two event persistence owners need coverage, not just the post-commit live event: +`chat-native-progress-work.ts` uses `registerScopedDeadlineWork` and the shared +delivery coordinator. It reconstructs relevant issue scopes on startup, refreshes +only the affected issue/company after a committed change, and projects batches +of 100. A full insertion batch retains immediate work. Lock contention retains a +one-second retry; failure uses the shared backoff. A settled short pass disarms. +The shared worker preserves changes during execution, paused work during drain, +startup failures through resume, fair company dispatch and shutdown ownership. -1. `heartbeat-run-events.ts::appendHeartbeatRunEvent`: native ingestion uses its - transaction/deduplication path. `paperclip-control-plane-port.ts` wraps this - in an outer transaction with active-authority checks before and after append. - Intent must precede the event write and follow outer commit/rollback. Duplicate - replay must not create fresh projection work. Ordinary non-progress log events - must not acquire a transaction/XID merely for this feature. -2. `native-run-coordinator-store.ts::appendEvent`: directly inserts run events - after current-run and sequence/duplicate checks. It does not call the generic - append helper. Cover it explicitly in its existing transaction. +Projection retains issue-before-run SKIP LOCKED ordering. A skipped lock is a +concrete obligation only while the matching row still exists and the run is +native/running. Deleted or terminal candidates settle. Authority, latest event, +phase deduplication and final output are checked again inside projection. -Additional paths found: +The selector also repairs an existing pagination bug: JavaScript Date truncated +PostgreSQL microseconds, repeatedly selecting a full rejected page. Cursors now +retain PostgreSQL timestamp text and the full timestamp/run/conversation tuple. +The regression seeds 30 origin-rejected candidates at one fractional timestamp +and an eligible later event. A query-count guard bounds failure of the old code. -- `provider-trace-workspace-diff-reprojection.ts` appends - `workspace.diff.recorded` through the generic native-source path. Recovery can - therefore create a new eligible event independently of live runner ingress. -- `heartbeat/run-lifecycle.ts` uses the generic writer for dynamic event types. - Preserve its redaction/runtime-status responsibilities; do not project raw data. -- Direct insertion in `heartbeat/queue.ts` writes a lifecycle event; it is outside - the closed progress allowlist. Coordinator-store direct insertion is not. -- Other native helper callers currently append lifecycle, process, cleanup, or - credential events. Their existence is not permission to signal every log entry. +## Evidence and failures retained -Eligibility changes can make a saved event actionable without a new event: -run/issue binding and state, conversation state, current identity/membership, -endpoint/connection/resource authority, lineage receipts, interaction arbitration, -and base-publication state. Reuse the existing issue/company-scoped notification -boundary audited in slice 28. Extend that shared boundary once; avoid repeated -recovery wrappers across routes. Inspect all base-publication retry and operator -resolution owners before removing the poll. `notifyChatPublicationWork` already -accepts an optional issue scope for projection-affecting changes; transport claim -bookkeeping intentionally does not invalidate projection. +- `/tmp/sleep29-progress-cursor-before.log`: causal pagination failure. +- `/tmp/sleep29-progress-cursor-acceptance.log`: 17 selected cases after repair. +- `/tmp/sleep29-progress-owner-acceptance.log`: 21 projection cases after extraction. +- `/tmp/sleep29-native-producers-before.log`: real event-writer cases fail before + notification hooks; startup and lock cases already pass. +- `/tmp/sleep29-native-producers-fixed.log`: all five original producer cases pass. +- `/tmp/sleep29-native-lifecycle.log`: 14 worker lifecycle cases pass. +- `/tmp/sleep29-progress-recovery.log`: trace recovery and 105-run backlog pass. + Its attempted publication-mode test was invalid: endpoint update does not + accept that field. No production API was expanded to satisfy the test. +- `/tmp/sleep29-progress-eligibility.log`: the first identity test incorrectly + assumed ordinary direct-origin projection enforced post-close authority. + The replacement uses the real closed/reopened conversation and admission + receipts, where the existing projection contract requires current authority. +- `/tmp/sleep29-progress-eligibility-fixed.log`: five real-service cases pass: + post-close identity, resource and visibility restoration, plus failed publication + replay and operator retry. No new event or manual projection call is used. +- `/tmp/sleep29-progress-worker-acceptance.log`: 13 app coordinator tests pass; + the concurrent-event test's quiet assertion initially counted old calls from + an already-installed transaction spy. Restoring that barrier spy before taking + the quiet baseline corrects the test accounting, without a production change. +- `/tmp/sleep29-final-native-acceptance.log`: 44 selected tests pass across the + 18 worker cases and 26 existing/new integration cases. These cover both real + event owners, nested outer commit/rollback, duplicate replay, irrelevant logs, + issue/run lock contention and terminalization, startup reconstruction, drain, + failed startup through resume, shutdown, trace recovery, issue/company scope, + concurrent commits, full batches, cadence, final suppression and provider-safe + content. The name filter skips the app file; its 13 cases passed separately. +- `/tmp/sleep29-wired-typecheck.log`: production server typecheck passes. +- `/tmp/sleep29-server-build.log`: server build passes (production TypeScript; + test files are executed by Vitest, not included in the server typecheck). +- `/tmp/sleep29-first-base-before.log`: causal missing-wake failure when a + first working publication commits after event selection. The test backdates + the base inside its transaction to model transaction-start timestamp ordering. +- `/tmp/sleep29-base-fixed-workers.log`: all 19 worker and 13 app coordinator + cases pass after the lifecycle producer signals native progress. +- Browser acceptance: all 16 `agent-chat-projects.spec.ts` cases pass, log + `/tmp/sleep29-app-e2e.log`, fresh isolated database, port 32659. This run + precedes the final base-publication hook; a final run remains pending. -## Contention and scheduling +PostgreSQL is external and isolated because embedded PostgreSQL is unavailable +on this host. Worker fixtures clean their own companies. The reused chat fixture +DB retains history; tests stop workers and retire endpoints after assertions. +Configured-idle and retained-history quiet assertions observe select/transaction +calls, including two minutes of controlled time in worker tests. Provider APIs +are controlled fixtures, not live Slack/Telegram/Discord accounts. -The current projector locks issue then run with SKIP LOCKED. This order avoids -inversion with native finalization and publication foreign keys. A zero insert -count currently conflates ordinary suppression, deletion/terminalization, and -lock contention. The polling caller retries all of them later. +## Limits and remaining acceptance -Before replacing that caller, expose whether concrete candidates were deferred -by contention. Retain a retry only for unresolved candidates or failure. Recheck -existence/current eligibility so a deleted or terminal run does not become an -endless retry. A later committed event or eligibility change must remain dirty -if it arrives during the current sweep. +Process-local notifications retain the single active writer model. Independent +SQL writers and other processes need a wake integration. No Cloud sleep gate or +external alarm coverage changes here. An internal deadline cannot wake stopped +compute. Browser acceptance, exact-head CI, and review remain required before +this slice is ready. The whole stack remains unmerged. -Use the shared scoped-deadline worker and delivery coordinator for coalescing, -company fairness, failure backoff, drain/resume and shutdown. Reconstruct relevant -issue scopes at startup. Scope refresh and projection must use the same candidate -contract. A short settled pass disarms; unfinished batches retain work. Preserve -24-hour policy rather than broadening history during this migration. - -## Required acceptance before removing periodic dispatch - -- Configured idle and retained completed/suppressed history generate no recurring - worker SQL after startup and settlement. -- Real event producers wake after outer commit, not savepoint release; rollback - publishes nothing and settles uncertainty through the existing protocol. -- Duplicate native-event replay and irrelevant log entries create no new debt. -- Both event persistence owners and trace reprojection have causal coverage. -- Restored authority/base-publication state wakes saved eligible events. -- Contended issue and run locks retain recovery while unrelated work proceeds; - terminalization, final answers and changed authority win before retry. -- Full batches, submillisecond cursor progression, cadence and phase deduplication. -- Restart, failure, changes during a pass, drain/resume, and shutdown. -- Assembled startup registers the worker and removes this one periodic lane; - other chat maintenance lanes remain until their own migrations. -- Current-head checks/review, final base reconciliation, and an accurate PR body. - -Process-local signals retain the single active writer model. No Cloud sleep gate -or external alarm coverage changes here; worker quiet is not provider suspension. - -## Remote reconciliation note - -Fetched master is `e302f2098f` (not merged into this stack yet). Newly landed +Fetched master is `19ec787e5b` (not replayed into this stack yet). Newly landed #17616 expands native-harness backup expiration to continuity/quarantine evidence -and adds its evidence signal. Its index.ts interval still exists. Audit this newer -contract when migrating that separate retention owner; do not recreate or drop -its preservation guarantees based on the old inventory. +and adds its evidence signal. Its interval remains. Preserve that newer contract +when migrating its separate retention owner. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 09c7ccc1abb..77cf0d1e5c8 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -25,7 +25,6 @@ describe("createChatReconciliationCoordinator", () => { processFailedGitHubWebhookDeliveries, processPendingDeliveries, flushPublications, - projectNativeProgress: async () => 0, processPendingSlackFileUploadReceipts: async () => undefined, processPendingSlackSessionSyncs: async () => undefined, onError, @@ -78,6 +77,8 @@ describe("createChatReconciliationCoordinator", () => { const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals")); expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); expect(maintenance).not.toContain("chatChannels.voice."); + expect(maintenance).not.toContain("enqueueSafeNativeChatProgress"); + expect(source).toContain("nativeChatProgressWork(db).register(deliveryWork,"); expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,"); const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8"); @@ -141,117 +142,6 @@ describe("createChatReconciliationCoordinator", () => { expect(actionMaintenance).not.toContain("processPendingGitHubWebhookIngress("); }); - it.each([false, true])( - "isolates a blocked native progress projector and joins its completion (rejects: %s)", - async (rejects) => { - let release!: () => void; - const held = new Promise((resolve) => { - release = resolve; - }); - const failure = new Error("native progress projection failed"); - const projectNativeProgress = vi - .fn(async () => 0) - .mockImplementationOnce(async () => { - await held; - if (rejects) throw failure; - return 1; - }); - const flushPublications = vi.fn(async () => undefined); - const onError = vi.fn(); - const coordinator = createChatReconciliationCoordinator({ - projectNativeProgress, - processPendingDeliveries: async () => undefined, - flushPublications, - processPendingSlackFileUploadReceipts: async () => undefined, - processPendingSlackSessionSyncs: async () => undefined, - onError, - }); - try { - coordinator.reconcile(); - await vi.waitFor(() => { - expect(projectNativeProgress).toHaveBeenCalledTimes(1); - expect(flushPublications).toHaveBeenCalledTimes(1); - }); - // A newly committed final/question must dispatch even while a different - // issue's projection is blocked. Repeated signals leave one dirty retry. - coordinator.notifyPublications(); - coordinator.notifyPublications(); - await vi.waitFor(() => - expect(flushPublications).toHaveBeenCalledTimes(2), - ); - expect(projectNativeProgress).toHaveBeenCalledTimes(1); - let drained = false; - const draining = coordinator.drain().then(() => { - drained = true; - }); - await new Promise((resolve) => setImmediate(resolve)); - expect(drained).toBe(false); - release(); - await draining; - expect(projectNativeProgress).toHaveBeenCalledTimes(2); - expect(onError).toHaveBeenCalledTimes(rejects ? 1 : 0); - if (rejects) - expect(onError).toHaveBeenCalledWith("native progress", failure); - const completedCounts = [ - projectNativeProgress.mock.calls.length, - flushPublications.mock.calls.length, - ]; - await new Promise((resolve) => setImmediate(resolve)); - expect([ - projectNativeProgress.mock.calls.length, - flushPublications.mock.calls.length, - ]).toEqual(completedCounts); - } finally { - release(); - coordinator.stop(); - await coordinator.drain(); - } - }, - ); - - it("joins a blocked native progress projector at shutdown without starting dirty follow-ups", async () => { - let release!: () => void; - const held = new Promise((resolve) => { - release = resolve; - }); - const projectNativeProgress = vi.fn(async () => { - await held; - return 1; - }); - const flushPublications = vi.fn(async () => undefined); - const coordinator = createChatReconciliationCoordinator({ - projectNativeProgress, - flushPublications, - processPendingDeliveries: async () => undefined, - processPendingSlackFileUploadReceipts: async () => undefined, - processPendingSlackSessionSyncs: async () => undefined, - onError: vi.fn(), - }); - try { - coordinator.reconcile(); - await vi.waitFor(() => { - expect(projectNativeProgress).toHaveBeenCalledOnce(); - expect(flushPublications).toHaveBeenCalledOnce(); - }); - coordinator.notifyPublications(); - coordinator.stop(); - let drained = false; - const draining = coordinator.drain().then(() => { - drained = true; - }); - await new Promise((resolve) => setImmediate(resolve)); - expect(drained).toBe(false); - release(); - await draining; - expect(projectNativeProgress).toHaveBeenCalledOnce(); - expect(flushPublications).toHaveBeenCalledOnce(); - } finally { - release(); - coordinator.stop(); - await coordinator.drain(); - } - }); - it("keeps slow optional recovery from suppressing later publication sweeps", async () => { let releaseDelivery!: () => void; let releaseSlackStatus!: () => void; @@ -269,7 +159,6 @@ describe("createChatReconciliationCoordinator", () => { ); const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ - projectNativeProgress: async () => 0, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, @@ -317,7 +206,6 @@ describe("createChatReconciliationCoordinator", () => { const processPendingSlackSessionSyncs = vi.fn(async () => undefined); const onError = vi.fn(); const coordinator = createChatReconciliationCoordinator({ - projectNativeProgress: async () => 0, processPendingDeliveries, flushPublications, processPendingSlackFileUploadReceipts, diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 85a1ee7205b..6e8d77a89c8 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,4 @@ +import { nativeChatProgressWork } from "../services/chat-native-progress-work.js"; import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "../middleware/auth.js"; import { cloudTenantPrimaryCompanyId } from "../services/cloud-instance.js"; import { accessService } from "../services/access.js"; @@ -52781,6 +52782,57 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }, ); + it.each(["identity", "resource", "visibility"] as const)( + "wakes saved post-close native progress when %s authority returns", async owner => { + const f = await reopenedFixture(); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const progress = () => db.select().from(chatPublications).where(and( + eq(chatPublications.issueId, f.conversation.issueId), like(chatPublications.idempotencyKey, "%:native:%"), + )); + try { + await db.insert(chatPublications).values({ + companyId: f.fixture.companyId, endpointId: f.endpoint.id, + conversationId: f.conversation.id, issueId: f.conversation.issueId, + idempotencyKey: `run:${f.fresh.runId}:working:${f.endpoint.id}`, + payload: { text: "Maya is working…", progressState: "working" }, state: "published", + createdAt: new Date(Date.now() - 60_000), + }); + await db.insert(heartbeatRunEvents).values({ + companyId: f.fixture.companyId, agentId: f.fixture.assignedAgentId, + runId: f.fresh.runId, seq: 1, eventType: "item.completed", + }); + let restore: () => Promise; + if (owner === "identity") { + await f.service.update(f.endpoint.id, { allowUnlinkedPeople: false }, "owner-user"); + await f.service.revokeLink(f.endpoint.id, f.fresh.action.principalId!); + const intent = await f.service.createLinkIntent(f.endpoint.id, f.fresh.action.principalId!, 1800); + restore = () => f.service.confirmIdentityLink(new URL(intent.confirmationUrl).searchParams.get("token")!, "owner-user"); + } else if (owner === "resource") { + const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.id, f.conversation.id)); + expect(conversation.resourceId).not.toBeNull(); + await f.service.replaceResources(f.endpoint.id, [{ id: conversation.resourceId!, enabled: false }], "owner-user"); + restore = () => f.service.replaceResources(f.endpoint.id, [{ id: conversation.resourceId!, enabled: true }], "owner-user"); + } else { + await issueService(db).update(f.conversation.issueId, { hiddenAt: new Date() }); + restore = () => issueService(db).update(f.conversation.issueId, { hiddenAt: null }); + } + await nativeChatProgressWork(db).register(coordinator, () => true).ready; + await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + expect(await progress()).toEqual([]); + await restore(); + await expect.poll(progress, { timeout: 5000 }).toEqual([ + expect.objectContaining({ idempotencyKey: `run:${f.fresh.runId}:working:${f.endpoint.id}:native:making_progress:1` }), + ]); + await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + const select = vi.spyOn(db, "select"); + try { await new Promise(resolve => setTimeout(resolve, 1200)); expect(select).not.toHaveBeenCalled(); } + finally { select.mockRestore(); } + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); await f.close(); } + }, + ); + it("publishes working and the exact fresh final after pinned close without resurrecting the old source", async () => { const f = await reopenedFixture(); try { @@ -63567,6 +63619,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it.each(["replay", "operator retry"] as const)( + "wakes saved native progress when %s restores eligibility", async owner => { + const context = await safeNativeProgressFixture("telegram", `12${["replay", "operator retry"].indexOf(owner)}`, "channel", "42", true); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const progress = () => db.select().from(chatPublications).where(and( + eq(chatPublications.issueId, context.conversation.issueId), + like(chatPublications.idempotencyKey, "%:native:%"), + )); + const settled = () => vi.waitFor(() => { + expect(coordinator.nextWakeAt()).toBeNull(); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + try { + const run = await context.createRun("saved progress awaiting eligibility"); + await context.addEvent(run); + const [base] = await db.update(chatPublications).set({ + state: owner === "replay" ? "failed" : "delivery_unknown", + }).where(eq(chatPublications.idempotencyKey, `run:${run.runId}:working:${context.endpoint.id}`)).returning(); + const restore = () => owner === "replay" + ? context.service.replayPublication(context.endpoint.id, base.id) + : context.service.resolvePublication(context.endpoint.id, base.id, "retry_anyway", "owner-user"); + await nativeChatProgressWork(db).register(coordinator, () => true).ready; + await settled(); expect(await progress()).toEqual([]); + await restore(); + await expect.poll(progress, { timeout: 5000 }).toEqual([ + expect.objectContaining({ idempotencyKey: `run:${run.runId}:working:${context.endpoint.id}:native:making_progress:1` }), + ]); + await settled(); + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await new Promise(resolve => setTimeout(resolve, 1200)); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); await retirePublicationFixture(context.service, context.endpoint.id); } + }, + ); + it("advances native progress past suppressed candidates with microsecond timestamps", async () => { const context = await safeNativeProgressFixture("telegram", "119"); try { diff --git a/server/src/__tests__/chat-native-progress-work.test.ts b/server/src/__tests__/chat-native-progress-work.test.ts new file mode 100644 index 00000000000..7d608be9fdc --- /dev/null +++ b/server/src/__tests__/chat-native-progress-work.test.ts @@ -0,0 +1,350 @@ +import { enqueueChatRunLifecycleMilestones } from "../services/chat-run-publications.js"; +import { persistReprojectedWorkspaceDiffs, projectCodexWorkspaceDiffsFromTrace } from "../services/provider-trace-workspace-diff-reprojection.js"; +import { randomUUID } from "node:crypto"; +import { eq, like, and } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { chatPublications, createDb, heartbeatRunEvents, heartbeatRuns, issues, subscribeDatabaseWork } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createChatMilestoneFixtures } from "./helpers/chat-milestone-fixtures.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { nativeChatProgressWork } from "../services/chat-native-progress-work.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; +import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; +import type { PrpEvent } from "../vendor/paperclip-runner/index.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_PROGRESS_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("native chat progress work", () => { + let db: ReturnType; + let temporary: Awaited> | undefined; + let fixtures: ReturnType; + let f: Awaited>; + let runId: string; + let coordinator: ReturnType | undefined; + let canRun = true; + const errors: unknown[] = []; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-chat-progress-"); + db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + fixtures = createChatMilestoneFixtures(db); + }, 60_000); + beforeEach(async () => { + errors.length = 0; canRun = true; f = await fixtures.create(); runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "running", responsibleUserId: f.userId, + contextSnapshot: { issueId: f.issueId, source: "chat:slack", endpointId: f.endpointId, wakeCommentId: f.commentId }, + }); + await db.insert(chatPublications).values({ companyId: f.companyId, endpointId: f.endpointId, + conversationId: f.conversationId, issueId: f.issueId, idempotencyKey: `run:${runId}:working:${f.endpointId}`, + payload: { text: "Maya is working…", progressState: "working" }, state: "published", + createdAt: new Date(Date.now() - 60_000), + }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); await fixtures.cleanup(); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + const publications = () => db.select().from(chatPublications).where(and( + eq(chatPublications.issueId, f.issueId), like(chatPublications.idempotencyKey, "%:native:%"), + )); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true, + onError: error => errors.push(error) }); + await nativeChatProgressWork(db).register(coordinator, () => canRun).ready; + } + async function settle() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + } + async function quiet() { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); } + finally { select.mockRestore(); transaction.mockRestore(); } + } + function event(nativeSource = false, eventType = "item.completed") { + return { companyId: f.companyId, agentId: f.agentId, runId, eventType, + message: "PRIVATE event text", payload: { token: "PRIVATE token" }, + ...(nativeSource ? { nativeSource: { sourceInstanceId: runId, sourceEventId: "event-1", sourceSeq: 1, + protocolSchemaVersion: 1, canonicalPayload: { eventType } } } : {}), + }; + } + + it.each([false, true])("wakes from a real committed event after idle (native source: %s)", async nativeSource => { + await start(); await settle(); await quiet(); expect(await publications()).toEqual([]); + const input = event(nativeSource); + const save = (rollback: boolean) => db.transaction(async tx => { + await appendHeartbeatRunEvent(tx as unknown as typeof db, input); + await vi.advanceTimersByTimeAsync(1000); + expect(await publications()).toEqual([]); + if (rollback) throw new Error("outer rollback"); + }); + await expect(save(true)).rejects.toThrow("outer rollback"); + await settle(); expect(await publications()).toEqual([]); await quiet(); + await save(false); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await publications()).toEqual([expect.objectContaining({ payload: expect.objectContaining({ text: "Maya is making progress…" }) })]); + }, { timeout: 5000 }); + await settle(); await quiet(); expect(errors).toEqual([]); + if (nativeSource) { + const signals: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind)); + try { expect((await appendHeartbeatRunEvent(db, input)).disposition).toBe("duplicate"); expect(signals).toEqual([]); } + finally { unsubscribe(); } + await quiet(); + } + }); + + it("wakes a saved event when its first working publication commits after event selection", async () => { + await db.delete(chatPublications).where(eq(chatPublications.issueId, f.issueId)); + await appendHeartbeatRunEvent(db, event(true)); + await start(); await settle(); expect(await publications()).toEqual([]); await quiet(); + await db.transaction(async tx => { + expect(await enqueueChatRunLifecycleMilestones(tx as unknown as typeof db, { + scope: { companyId: f.companyId, issueId: f.issueId }, + })).toBe(1); + // Model a base transaction that began before the event but committed + // afterwards, without a wall-clock delay in the suite. PostgreSQL's + // default now() is transaction-start time, not commit time. + await tx.update(chatPublications).set({ createdAt: new Date(Date.now() - 300_000) }) + .where(eq(chatPublications.idempotencyKey, `run:${runId}:working:${f.endpointId}`)); + await vi.advanceTimersByTimeAsync(1000); expect(await publications()).toEqual([]); + }); + await settle(); expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]); + }); + + it("reconstructs saved progress after restart and becomes quiet with published history", async () => { + await appendHeartbeatRunEvent(db, event(true)); + await start(); await settle(); expect(await publications()).toHaveLength(1); await quiet(); + await coordinator!.stop(); await start(); await settle(); + expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]); + }); + + it.each(["issue", "run"] as const)("retries a contended %s without another event", async owner => { + await appendHeartbeatRunEvent(db, event(true)); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const lock = db.transaction(async tx => { + if (owner === "issue") await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); + else await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update"); + entered(); await hold; + }); + try { + await ready; await start(); expect(await publications()).toEqual([]); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + release(); await lock; + await settle(); expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]); + } finally { release(); await lock; } + }); + + it.each(["port", "coordinator"] as const)("wakes from the real %s event owner after outer commit, with quiet replay", async kind => { + const sessionId = randomUUID(), runnerId = randomUUID(); + await db.update(heartbeatRuns).set({ nativeSessionId: sessionId }).where(eq(heartbeatRuns.id, runId)); + const binding = { companyId: f.companyId, issueId: f.issueId, runId, agentId: f.agentId, + sessionId, completionContractId: randomUUID(), completionContractSha256: "fixture", + sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "control", + normalizedSessionId: sessionId, runnerSourceInstanceId: runnerId, + completionContractRevision: "1", completionContractCriterionIds: [], + }; + const source: PrpEvent = { schema: "paperclip.prp.event.v1", schemaVersion: 1, + sourceEventId: "event-1", sourceSeq: 1, sourceInstanceId: runnerId, sourceKind: "runner", + runId, normalizedSessionId: sessionId, turnId: "turn-1", itemId: "item-1", + eventType: "item.completed", priority: 1, emittedAt: new Date().toISOString(), payload: {}, + }; + const append = (client: typeof db) => (kind === "port" + ? new PaperclipControlPlanePort(client, binding) + : new NativeRunCoordinatorStore(client, binding)).appendEvent(source); + await start(); await settle(); await quiet(); + await expect(db.transaction(async tx => { + await append(tx as unknown as typeof db); throw new Error("rollback owner"); + })).rejects.toThrow("rollback owner"); + await settle(); expect(await publications()).toEqual([]); await quiet(); + await db.transaction(async tx => { + await append(tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(1000); + expect(await publications()).toEqual([]); + }); + await settle(); expect(await publications()).toHaveLength(1); await quiet(); + const signals: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind)); + try { expect((await append(db)).disposition).toBe("duplicate"); expect(signals).toEqual([]); } + finally { unsubscribe(); } + await quiet(); expect(errors).toEqual([]); + }); + + it.each(["issue", "run"] as const)("disarms after terminalization wins during a contended %s", async owner => { + await appendHeartbeatRunEvent(db, event(true)); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const lock = db.transaction(async tx => { + if (owner === "issue") await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); + else await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update"); + entered(); await hold; + await tx.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, runId)); + }); + try { + await ready; await start(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + release(); await lock; await settle(); expect(await publications()).toEqual([]); await quiet(); + expect(errors).toEqual([]); + } finally { release(); await lock; } + }); + + it.each(["first write", "startup"] as const)("retains %s through drain without passive SQL", async kind => { + if (kind === "first write") { await start(); await settle(); } + canRun = false; + await appendHeartbeatRunEvent(db, event(true)); + if (kind === "startup") await start(); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(await publications()).toEqual([]); await quiet(); + canRun = true; await vi.advanceTimersByTimeAsync(30_000); await settle(); + expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]); + }); + + it("retains failed startup selection through drain and retries after resume", async () => { + await appendHeartbeatRunEvent(db, event(true)); + const select = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("temporary read failure"); }); + try { await start(); } finally { select.mockRestore(); } + expect(errors.splice(0)).toHaveLength(1); canRun = false; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + await quiet(); expect(await publications()).toEqual([]); + canRun = true; await vi.advanceTimersByTimeAsync(30_000); await settle(); + expect(await publications()).toHaveLength(1); await quiet(); expect(errors).toEqual([]); + }); + + it("leaves unrelated direct log events on their atomic statement path without projection intent", async () => { + await start(); await settle(); + const signals: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind)); + const transaction = vi.spyOn(db, "transaction"); + try { + await appendHeartbeatRunEvent(db, event(false, "item.delta")); + expect(transaction).not.toHaveBeenCalled(); expect(signals).toEqual([]); + } finally { transaction.mockRestore(); unsubscribe(); } + await quiet(); expect(await publications()).toEqual([]); expect(errors).toEqual([]); + }); + + it("wakes saved trace reprojection without live runner ingress and leaves duplicate recovery quiet", async () => { + await start(); await settle(); await quiet(); + const projection = projectCodexWorkspaceDiffsFromTrace([{ + kind: "frame", direction: "provider_to_client", frameId: 1, + rawBase64: Buffer.from(JSON.stringify({ method: "turn/diff/updated", params: { + turnId: "turn-1", diff: "diff --git a/private.txt b/private.txt\n--- a/private.txt\n+++ b/private.txt\n@@ -1 +1 @@\n-private\n+PRIVATE replacement\n", + } })).toString("base64"), + }]); + expect(projection.turns).toHaveLength(1); + const input = { traceId: randomUUID(), companyId: f.companyId, agentId: f.agentId, runId, projection }; + expect(await persistReprojectedWorkspaceDiffs(db, input)).toMatchObject({ created: 1 }); + await settle(); + expect(await publications()).toEqual([expect.objectContaining({ payload: expect.objectContaining({ text: "Maya is working with files…" }) })]); + await quiet(); + const signals: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatNativeProgress, kind => signals.push(kind)); + try { expect(await persistReprojectedWorkspaceDiffs(db, input)).toMatchObject({ created: 0 }); expect(signals).toEqual([]); } + finally { unsubscribe(); } + await quiet(); expect(errors).toEqual([]); + }); + + it("retains a committed event that arrives before the current projection pass returns", async () => { + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const secondRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: secondRunId, companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.issueId, + runtimeMode: "native", status: "running", responsibleUserId: f.userId, contextSnapshot: source.contextSnapshot, + }); + await db.insert(chatPublications).values({ + companyId: f.companyId, endpointId: f.endpointId, conversationId: f.conversationId, + issueId: f.issueId, idempotencyKey: `run:${secondRunId}:working:${f.endpointId}`, + payload: { text: "Maya is working…", progressState: "working" }, state: "published", + createdAt: new Date(Date.now() - 60_000), + }); + await start(); await settle(); + let held = false, release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + const transaction = db.transaction.bind(db); + const barrier = vi.spyOn(db, "transaction").mockImplementation(async (callback, options) => { + const result = await transaction(callback, options); + // Pause after the first projection commits, before its pass can disarm. + // Producer transactions return event records, not an inserted row count. + if (result === 1 && !held) { held = true; await pending; } + return result; + }); + try { + await appendHeartbeatRunEvent(db, event(true)); + await vi.waitFor(() => expect(held).toBe(true), { timeout: 5000 }); + await appendHeartbeatRunEvent(db, { ...event(false), runId: secondRunId }); + release(); await settle(); + expect(await publications()).toHaveLength(2); barrier.mockRestore(); + await quiet(); expect(errors).toEqual([]); + } finally { release(); barrier.mockRestore(); } + }); + + it("limits an event wake to its issue and reconstructs other saved scopes on restart", async () => { + await start(); await settle(); + // Seed rows without signals to expose an accidentally global event wake. + const others = [await fixtures.create(f.companyId), await fixtures.create()]; + for (const other of others) { + const id = randomUUID(); + await db.insert(heartbeatRuns).values({ + id, companyId: other.companyId, agentId: other.agentId, nativeIssueId: other.issueId, + runtimeMode: "native", status: "running", responsibleUserId: other.userId, + contextSnapshot: { issueId: other.issueId, source: "chat:slack", endpointId: other.endpointId, wakeCommentId: other.commentId }, + }); + await db.insert(chatPublications).values({ + companyId: other.companyId, endpointId: other.endpointId, conversationId: other.conversationId, + issueId: other.issueId, idempotencyKey: `run:${id}:working:${other.endpointId}`, + payload: { text: "Maya is working…", progressState: "working" }, state: "published", + createdAt: new Date(Date.now() - 60_000), + }); + await db.insert(heartbeatRunEvents).values({ companyId: other.companyId, agentId: other.agentId, + runId: id, seq: 1, eventType: "item.completed" }); + } + const saved = (issueId: string) => db.select().from(chatPublications).where(and( + eq(chatPublications.issueId, issueId), like(chatPublications.idempotencyKey, "%:native:%"), + )); + await appendHeartbeatRunEvent(db, event(true)); await settle(); + expect(await publications()).toHaveLength(1); + for (const other of others) expect(await saved(other.issueId)).toEqual([]); + await quiet(); await coordinator!.stop(); await start(); await settle(); + for (const other of others) expect(await saved(other.issueId)).toHaveLength(1); + await quiet(); expect(errors).toEqual([]); + }); + + it("drains more than one issue batch and becomes quiet with all progress retained", async () => { + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const ids = Array.from({ length: 105 }, () => randomUUID()); + await db.insert(heartbeatRuns).values(ids.map(id => ({ + id, companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.issueId, + runtimeMode: "native", status: "running", responsibleUserId: f.userId, contextSnapshot: source.contextSnapshot, + }))); + await db.insert(chatPublications).values(ids.map(id => ({ + companyId: f.companyId, endpointId: f.endpointId, conversationId: f.conversationId, + issueId: f.issueId, idempotencyKey: `run:${id}:working:${f.endpointId}`, + payload: { text: "Maya is working…", progressState: "working" }, state: "published", + createdAt: new Date(Date.now() - 60_000), + }))); + await db.insert(heartbeatRunEvents).values(ids.map(id => ({ + companyId: f.companyId, agentId: f.agentId, runId: id, seq: 1, eventType: "item.completed", + }))); + await start(); await settle(); + expect(await publications()).toHaveLength(ids.length); await quiet(); expect(errors).toEqual([]); + }); + + it("unsubscribes at shutdown and reconstructs a later event on restart", async () => { + await start(); await settle(); await coordinator!.stop(); + await appendHeartbeatRunEvent(db, event(true)); + await quiet(); expect(await publications()).toEqual([]); + await start(); await settle(); expect(await publications()).toHaveLength(1); expect(errors).toEqual([]); + }); + +}); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index c367b3d6670..1ec0c06fb7c 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -173,7 +173,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id)); const onError = vi.fn(); const maintenance = createChatReconciliationCoordinator({ - processPendingDeliveries: async () => undefined, projectNativeProgress: async () => 0, + processPendingDeliveries: async () => undefined, flushPublications: async () => undefined, onError, }); try { diff --git a/server/src/app.ts b/server/src/app.ts index 68377fb9d23..c2813d181b9 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -195,7 +195,7 @@ import { createToolGatewayService } from "./services/tool-gateway.js"; import { toolAccessService } from "./services/tool-access.js"; import { chatChannelService } from "./services/chat-channels.js"; import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js"; -import { enqueueSafeNativeChatProgress } from "./services/chat-run-publications.js"; +import { nativeChatProgressWork } from "./services/chat-native-progress-work.js"; import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js"; import { createCoalescedAsyncTrigger, @@ -324,7 +324,6 @@ export function shouldEnablePrivateHostnameGuard(opts: { type ChatReconciliationLane = | "deliveries" | "GitHub webhook recovery" - | "native progress" | "publications" | "Slack file receipts" | "Slack session status"; @@ -337,7 +336,6 @@ type ChatReconciliationLane = export function createChatReconciliationCoordinator(input: { processPendingDeliveries: () => Promise; processFailedGitHubWebhookDeliveries?: () => Promise; - projectNativeProgress: () => Promise; flushPublications: () => Promise; processPendingSlackFileUploadReceipts?: () => Promise; processPendingSlackSessionSyncs?: () => Promise; @@ -349,16 +347,6 @@ export function createChatReconciliationCoordinator(input: { run: input.flushPublications, onError: (error) => input.onError("publications", error), }); - const progressReconciliation = createCoalescedAsyncTrigger({ - run: async () => { - const inserted = await input.projectNativeProgress(); - // Existing final/question publications never wait on this optional - // projection. Newly committed progress gets a bounded dispatch wake; - // an empty/contended pass does not create a self-sustaining loop. - if (inserted > 0) publicationReconciliation.notify(); - }, - onError: (error) => input.onError("native progress", error), - }); const start = ( lane: ChatReconciliationLane, task: () => Promise, @@ -383,27 +371,21 @@ export function createChatReconciliationCoordinator(input: { input.processFailedGitHubWebhookDeliveries, ); } - progressReconciliation.poll(); publicationReconciliation.poll(); if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts); if (input.processPendingSlackSessionSyncs) start("Slack session status", input.processPendingSlackSessionSyncs); }, notifyPublications() { - progressReconciliation.notify(); publicationReconciliation.notify(); }, stop() { stopped = true; - progressReconciliation.stop(); publicationReconciliation.stop(); }, async drain() { await Promise.allSettled([ ...inFlight.values(), - progressReconciliation.drain(), ]); - // Projecting the final batch can notify dispatch after an earlier drain - // would have returned. Join dispatch only after its producer has drained. await publicationReconciliation.drain(); }, }; @@ -1305,6 +1287,7 @@ export async function createApp( chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatRunMilestoneWork(db, opts.authPublicBaseUrl).register( deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(), ); @@ -1318,7 +1301,6 @@ export async function createApp( processPendingDeliveries: async () => undefined, processFailedGitHubWebhookDeliveries: () => chatChannels.processFailedGitHubWebhookDeliveries(), - projectNativeProgress: () => enqueueSafeNativeChatProgress(db), flushPublications: () => reconcileChatPublicationMaintenance(), onError: (lane, err) => { logger.error({ err, lane }, `Failed to reconcile chat ${lane}`); @@ -1330,8 +1312,8 @@ export async function createApp( chatReconciliation.notifyPublications(); }, ); - // Native progress and remaining publication maintenance still use this cadence. - // Lifecycle milestones use committed changes and their own recovery worker. + // Remaining publication maintenance still uses this cadence. Lifecycle and + // native progress projection have independent commit/deadline workers. // Inbox, outbox, and file receipt queues have independent commit/deadline wakes. let chatPublicationTimer: ReturnType | null = setInterval( () => { diff --git a/server/src/services/chat-channels/publications/resolution.ts b/server/src/services/chat-channels/publications/resolution.ts index b782e01d184..2b54f9bd2c9 100644 --- a/server/src/services/chat-channels/publications/resolution.ts +++ b/server/src/services/chat-channels/publications/resolution.ts @@ -113,7 +113,7 @@ export function createChatPublicationResolution(db: Db, options: ResolutionOptio code: "chat_file_transfer_resolution_required", }, ); - await notifyChatPublicationWork(tx); + await notifyChatPublicationWork(tx, current); return tx .update(chatPublications) .set({ @@ -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, action === "retry_anyway" ? undefined : publication); + await notifyChatPublicationWork(tx, publication); await notifyVoicePublication(tx, initialRecord.endpoint); await tx .update(chatPublications) diff --git a/server/src/services/chat-native-progress-notifications.ts b/server/src/services/chat-native-progress-notifications.ts new file mode 100644 index 00000000000..742e5284d2d --- /dev/null +++ b/server/src/services/chat-native-progress-notifications.ts @@ -0,0 +1,6 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import type { ChatRunProjectionScope } from "./chat-run-publication-policy.js"; + +export function notifyChatNativeProgress(tx: object, scope: ChatRunProjectionScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatNativeProgress, JSON.stringify(scope)); +} diff --git a/server/src/services/chat-native-progress-work.ts b/server/src/services/chat-native-progress-work.ts new file mode 100644 index 00000000000..62abd3c3480 --- /dev/null +++ b/server/src/services/chat-native-progress-work.ts @@ -0,0 +1,34 @@ +import type { Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; +import type { ChatRunProjectionScope } from "./chat-run-publication-policy.js"; +import { loadNativeChatProgressScopes, reconcileNativeChatProgress } from "./chat-native-progress.js"; + +type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; +const batchSize = 100; + +export function nativeChatProgressWork(db: Db) { + async function load(scope?: ChatRunProjectionScope, signal?: AbortSignal): Promise { + return (await loadNativeChatProgressScopes(db, scope, signal)).map(row => ({ + ...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0, + })); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + const result = await reconcileNativeChatProgress(db, { + scope: { companyId: entry.companyId, issueId: entry.issueId }, limit: batchSize, signal, + }); + if (result.inserted === batchSize) return { ...entry, dueAt: Date.now() }; + return result.contended ? { ...entry, dueAt: Date.now() + 1000 } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatNativeProgress, + decodeScope: (hint: string): ChatRunProjectionScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: ChatRunProjectionScope) => !scope || + (entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId)), + load, check, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/chat-native-progress.ts b/server/src/services/chat-native-progress.ts index 47b87e6a2ee..426cedfb1aa 100644 --- a/server/src/services/chat-native-progress.ts +++ b/server/src/services/chat-native-progress.ts @@ -7,7 +7,7 @@ import { projectSafeChatPublication } from "./chat-publication-projection.js"; import { hasChatRunOwnedProviderInteraction } from "./chat-interaction-arbitration.js"; import { resolveChatOriginPublicationBindings } from "./issues.js"; import { boundRunIssueIdSql } from "./run-task-binding.js"; -import { allowsAutomaticRunPublication, chatRunProjectionWindow } from "./chat-run-publication-policy.js"; +import { allowsAutomaticRunPublication, chatRunProjectionWindow, type ChatRunProjectionScope } from "./chat-run-publication-policy.js"; import { SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, safeNativeChatProgressForEvent } from "./safe-native-chat-progress.js"; type SafeNativeChatProgressCandidate = { @@ -26,12 +26,10 @@ type SafeNativeChatProgressCandidate = { const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000; -/** Only closed event types become coarse progress; event payloads stay local. */ -export async function enqueueSafeNativeChatProgress( - db: Db, - options: { since?: Date; limit?: number } = {}, -): Promise { - const input = chatRunProjectionWindow(options); +type NativeProgressCursor = Pick; +type NativeProgressInput = { since?: Date; limit?: number; scope?: ChatRunProjectionScope; signal?: AbortSignal }; + +function nativeProgressPages(db: Db, since: Date, scope?: ChatRunProjectionScope) { const basePublication = alias( chatPublications, "safe_native_progress_base_publication", @@ -41,17 +39,7 @@ export async function enqueueSafeNativeChatProgress( "safe_native_progress_later_event", ); const issueIdFromContext = boundRunIssueIdSql(); - let inserted = 0; - let cursor: { - eventCreatedAt: string; - runId: string; - conversationId: string; - } | null = null; - - while (inserted < input.limit) { - const pageSize = Math.max(25, Math.min(200, input.limit - inserted)); - const pageCursor: typeof cursor = cursor; - const rows: SafeNativeChatProgressCandidate[] = await db + return async (pageSize: number, pageCursor: NativeProgressCursor | null): Promise => await db .select({ eventId: heartbeatRunEvents.id, eventSeq: heartbeatRunEvents.seq, @@ -118,7 +106,9 @@ export async function enqueueSafeNativeChatProgress( inArray(heartbeatRunEvents.eventType, [ ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, ]), - gte(heartbeatRunEvents.createdAt, input.since), + gte(heartbeatRunEvents.createdAt, since), + scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(chatConversations.issueId, scope.issueId) : undefined, sql`${heartbeatRunEvents.createdAt} >= ${basePublication.createdAt} + interval '20 seconds'`, notExists( db @@ -147,6 +137,31 @@ export async function enqueueSafeNativeChatProgress( asc(chatConversations.id), ) .limit(pageSize); +} + +export async function loadNativeChatProgressScopes(db: Db, scope?: ChatRunProjectionScope, signal?: AbortSignal) { + const pages = nativeProgressPages(db, chatRunProjectionWindow({}).since, scope); + const scopes = new Map(); + let cursor: NativeProgressCursor | null = null; + for (;;) { + signal?.throwIfAborted(); + const rows = await pages(200, cursor); + for (const row of rows) scopes.set(`${row.companyId}:${row.issueId}`, { companyId: row.companyId, issueId: row.issueId }); + if (rows.length < 200) return [...scopes.values()]; + cursor = rows[rows.length - 1]; + } +} + +/** A suppressed event is settled; lock contention still owes another attempt. */ +export async function reconcileNativeChatProgress(db: Db, options: NativeProgressInput = {}) { + const input = chatRunProjectionWindow(options); + const pages = nativeProgressPages(db, input.since, options.scope); + let inserted = 0, contended = false; + let cursor: NativeProgressCursor | null = null; + while (inserted < input.limit) { + options.signal?.throwIfAborted(); + const pageSize = Math.max(25, Math.min(200, input.limit - inserted)); + const rows = await pages(pageSize, cursor); if (rows.length === 0) break; const lastRow = rows.at(-1)!; cursor = { @@ -156,6 +171,7 @@ export async function enqueueSafeNativeChatProgress( }; for (const row of rows) { + options.signal?.throwIfAborted(); if (inserted >= input.limit) break; const result = await db.transaction(async (tx) => { // Keep the normal issue -> run order used by native finalization. The @@ -174,7 +190,12 @@ export async function enqueueSafeNativeChatProgress( ) .for("update", { skipLocked: true }) .then((currentRows) => currentRows[0] ?? null); - if (!currentIssue) return 0; + if (!currentIssue) { + const [existing] = await tx.select({ id: issues.id }).from(issues) + .where(and(eq(issues.id, row.issueId), eq(issues.companyId, row.companyId))); + contended ||= Boolean(existing); + return 0; + } // Native event admission serializes on the run row. Taking the same // short lock makes terminal status, a later event, or a final comment // win before this non-terminal projection can be inserted. @@ -192,7 +213,15 @@ export async function enqueueSafeNativeChatProgress( ) .for("update", { skipLocked: true }) .then((currentRows) => currentRows[0] ?? null); - if (!currentRun) return 0; + if (!currentRun) { + const [existing] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, row.runId), eq(heartbeatRuns.companyId, row.companyId), + eq(heartbeatRuns.agentId, row.agentId), eq(heartbeatRuns.runtimeMode, "native"), + eq(heartbeatRuns.status, "running"), + )); + contended ||= Boolean(existing); + return 0; + } const destination = await tx .select({ id: chatConversations.id }) @@ -389,5 +418,10 @@ export async function enqueueSafeNativeChatProgress( } if (rows.length < pageSize) break; } - return inserted; + return { inserted, contended }; +} + +/** Explicit/manual callers retain the original count-only contract. */ +export async function enqueueSafeNativeChatProgress(db: Db, options: NativeProgressInput = {}): Promise { + return (await reconcileNativeChatProgress(db, options)).inserted; } diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 743a7dd992f..d3f38e52d81 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,9 +1,13 @@ +import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; /** 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 function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { - return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); + // Lifecycle and coarse native progress share lineage and authority. Extend + // this eligibility boundary once rather than adding wrappers to its callers. + await notifyChatNativeProgress(tx, scope); } diff --git a/server/src/services/chat-run-publication-policy.ts b/server/src/services/chat-run-publication-policy.ts index 4fa5eed3879..0ce57ff5794 100644 --- a/server/src/services/chat-run-publication-policy.ts +++ b/server/src/services/chat-run-publication-policy.ts @@ -17,3 +17,5 @@ export function chatRunProjectionWindow(input: { since?: Date; limit?: number }) limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)), }; } + +export type ChatRunProjectionScope = { companyId: string; issueId?: string }; diff --git a/server/src/services/chat-run-publications.ts b/server/src/services/chat-run-publications.ts index 8238cbeb41c..a4d59aab455 100644 --- a/server/src/services/chat-run-publications.ts +++ b/server/src/services/chat-run-publications.ts @@ -1,3 +1,4 @@ +import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; import { allowsAutomaticRunPublication, chatRunProjectionWindow } from "./chat-run-publication-policy.js"; import { enqueueSafeNativeChatProgress } from "./chat-native-progress.js"; export { enqueueSafeNativeChatProgress } from "./chat-native-progress.js"; @@ -646,6 +647,9 @@ export async function enqueueChatRunLifecycleMilestones( if (explicitlyAuthoredPublication.length > 0) continue; } const result = await db.transaction(async (tx) => { + if (milestone === "working") { + await notifyChatNativeProgress(tx, { companyId: row.companyId, issueId: row.issueId }); + } await notifyChatPublicationWork(tx); return tx .insert(chatPublications) diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index f9b3f0f0c9a..951672618b3 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -54,6 +54,7 @@ export const DELIVERY_QUEUES = { chatVerification: "chat-verification", chatInbound: "chat-inbound", chatPublications: "chat-publications", + chatNativeProgress: "chat-native-progress", chatRunMilestones: "chat-run-milestones", chatReceipts: "chat-slack-receipts", browser: "browser-use", diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index ecb7ff5864a..c11687c9045 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -1,3 +1,6 @@ +import { isSafeNativeChatProgressEventType } from "./safe-native-chat-progress.js"; +import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; +import { boundRunIssueIdSql } from "./run-task-binding.js"; import { and, asc, eq, getTableColumns, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; @@ -68,7 +71,8 @@ export async function appendHeartbeatRunEvent( db: Db, input: AppendHeartbeatRunEventInput, ): Promise { - if (!input.nativeSource && !input.retryExhaustion) { + const projectsProgress = isSafeNativeChatProgressEventType(input.eventType); + if (!input.nativeSource && !input.retryExhaustion && !projectsProgress) { const allocated = db.$with("allocated_run_event", {}).as(sql` update ${heartbeatRuns} set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1, @@ -94,7 +98,8 @@ export async function appendHeartbeatRunEvent( } return db.transaction(async (tx) => { const run = await tx - .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) + .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + runtimeMode: heartbeatRuns.runtimeMode, issueId: boundRunIssueIdSql() }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .for("update") @@ -165,6 +170,9 @@ export async function appendHeartbeatRunEvent( } } + if (projectsProgress && run.runtimeMode === "native" && run.issueId) { + await notifyChatNativeProgress(tx, { companyId: run.companyId, issueId: run.issueId }); + } const seq = await allocateHeartbeatRunEventSeq( tx as unknown as Db, input.runId, diff --git a/server/src/services/native-runtime/native-run-coordinator-store.ts b/server/src/services/native-runtime/native-run-coordinator-store.ts index 99b80907c66..34e45437d2f 100644 --- a/server/src/services/native-runtime/native-run-coordinator-store.ts +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -1,3 +1,5 @@ +import { isSafeNativeChatProgressEventType } from "../safe-native-chat-progress.js"; +import { notifyChatNativeProgress } from "../chat-native-progress-notifications.js"; import { createHash } from "node:crypto"; import { and, desc, eq, inArray, lt } from "drizzle-orm"; @@ -290,6 +292,9 @@ export class NativeRunCoordinatorStore { throw new Error("native_event_source_gap"); } + if (isSafeNativeChatProgressEventType(event.eventType)) { + await notifyChatNativeProgress(tx, { companyId: this.#binding.companyId, issueId: this.#binding.issueId }); + } const cursor = run.nextEventSeq; const [inserted] = await tx .insert(heartbeatRunEvents)