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 6d3ea3beff1..a756578e9c1 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2170,3 +2170,21 @@ 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 + +PR [#17639](https://github.com/paperclipai/paperclip/pull/17639), 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. A second fresh browser run passes all +16 cases at `bcf2f38ff5`; final typecheck passes too. Remote checks/review 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 new file mode 100644 index 00000000000..dc4ab31df5e --- /dev/null +++ b/doc/plans/2026-10-25-native-chat-progress-sleep-audit.md @@ -0,0 +1,225 @@ +# Native chat progress: sleep migration audit + +Parent: #17606 at `ed76256b979541b962fa784016a550c32920d670`. +Branch: `codex/sleep-29-native-chat-progress`. PR: #17639. +Status: implementation, focused PostgreSQL acceptance, and browser acceptance +complete; exact-head remote checks and review pending. This is not a deployed sleep claim. + +## Duty and owner + +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, authorization and insertion. +`safe-native-chat-progress.ts` owns the normalized allowlist and phrase policy. +`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. + +## Durable contract and triggers + +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. + +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 two event writers register awaited intent before insertion, after duplicate +and ownership checks, inside their existing transaction boundaries: + +- `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. + +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. + +## Scheduling, contention and lifetime + +`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. + +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. + +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. + +## Evidence and failures retained + +- `/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. `/tmp/sleep29-final-e2e.log` then + passes all 16 cases at committed implementation `bcf2f38ff5` with a second + fresh isolated database. `/tmp/sleep29-final-typecheck.log` also passes with + the final hook. No retries were needed in either browser run. + +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. + +## Limits and remaining acceptance + +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. Exact-head CI and review remain required before this slice is ready. The whole stack remains unmerged. + +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 interval remains. Preserve that newer contract +when migrating its separate retention owner. + +## CI repair: narrow notification payload (2026-10-25) + +CI at `34df8b0d01` failed native status/finalization consumers across server, +chat, Runner, and serialized suites. Local reproduction of the existing +same-run Slack response-recovery test failed before the repair. Its persisted +`native_run_finalizations.failure_detail.message` identified circular JSON +serialization of a Drizzle table through the caller's database handle. + +The new native-progress notifier serialized the entire structurally compatible +scope argument. Existing lifecycle callers legitimately pass wider domain inputs +(including `db`) through that typed boundary. Serialize only `companyId` and +`issueId`, as the lifecycle notifier already does; do not constrain every caller +or serialize domain/private fields into work hints. + +After repair, 15 existing chat finalization/response-recovery/authority cases +passed against external PostgreSQL, all 32 native-progress worker/startup cases +passed, and server typecheck passed. The attempted broader native-finalization +and session-resumption selection failed during local embedded PostgreSQL +initialization (bootstrap exit 1), before tests could run; it is not acceptance +evidence. Fresh CI must verify those native consumers. Before/after logs: +`/tmp/sleep29-finalization-repro.log`, `/tmp/sleep29-finalization-fixed.log`, +`/tmp/sleep29-notification-fix-workers.log`, +`/tmp/sleep29-notification-fix-typecheck.log`, and +`/tmp/sleep29-native-finalization-acceptance.log`. + +## CI repair: keep non-progress event admission narrow (2026-10-25) + +CI at `3447973f6f` clears the prior finalization failures. Its sole failing test +file is `heartbeat-run-event-atomic-append.test.ts`: native lifecycle and retry +exhaustion events unnecessarily select runtime mode and the task-binding SQL, +which can read a large context snapshot. Those events do not project progress. +The lock query now selects the extra binding only for allowlisted progress +kinds. Keep the existing strict SQL/payload regression assertions unchanged. + +The atomic-event suite now supports the repository's guarded external test DB +convention, allowing its actual PostgreSQL SQL assertions to run locally despite +embedded bootstrap failure. The original implementation reproduces both CI +failures in `/tmp/sleep29-atomic-events-before.log`. That first external run also +exposed fixture cleanup ordering; cleanup now removes event/run/agent rows before +companies and always closes its pools. The repaired atomic-event and progress +worker suites pass all 25 cases in `/tmp/sleep29-atomic-progress-fixed.log`; +server typecheck passes in `/tmp/sleep29-atomic-progress-typecheck.log`. + +Greptile reviewed `3447973f6f` at 5/5 with no threads. A new full CI/review pass +is required for this second repair; do not carry forward readiness from its +parent head. + + +## Current milestone-parent acceptance (October 28) + +The existing native-progress slice is prepared with a normal merge of the verified +local milestone parent `eef52b9dd759417bd7a2fea835357fbfeb460111`. Its previous +published head is `08d07ead105dc89c2af9b62f4d90bff8cc93540c`, with original own +base `7991c2fa2ce7298bee69b25dbc6cb60a5e2fe250`. All unrelated parent changes, +including the current migration 0397 and authority, voice, and identity repairs, +remain present. The only explicit-base conflict was publication resolution: keep +this slice's issue-scoped retry notification and the parent's exact voice +conversation scope. No new duty, table, dependency, or scheduler is introduced. + +Fresh separate PostgreSQL fixtures pass **163 distinct selected cases**: +**137 across eight complete suites** (native progress, atomic event append, +all five milestone worker suites, and app admission wiring), plus **26 real chat +integration cases** selected by `native progress`. These include replay/operator +retry, microsecond pagination, lock contention, current authority, cadence, phase +coalescing, final/question precedence, and redaction. The other **1,595 chat cases +were skipped**, not passed. Server production typechecking, build, and module +boundaries also pass; test files were executed rather than statically checked. + +Commands, separate fixture names, output, and zero exit-status receipts are in +`/tmp/sleep29-current-parent-acceptance-command.json` and +`/tmp/sleep29-current-parent-{acceptance,producers,types,build,boundaries}.log`. +Earlier verification and failures above keep their stated revision scope. This +is local preparation only: final landed-parent reconciliation, all fresh exact- +head remote gates, and Greptile 5/5 remain required before merge. This does not +qualify a live provider, external alarm, whole-instance suspension, or savings. 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 8428c0cb192..fff734e5b36 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"; @@ -53367,6 +53368,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 { @@ -65646,6 +65698,91 @@ 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 { + const valid = await context.createRun("eligible after rejected page"); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, valid.runId)); + const rejectedIds = Array.from({ length: 30 }, () => randomUUID()); + await db.insert(heartbeatRuns).values(rejectedIds.map(id => ({ + id, companyId: source.companyId, agentId: source.agentId, + runtimeMode: "native", status: "running", + contextSnapshot: { ...source.contextSnapshot, endpointId: randomUUID() }, + }))); + await db.insert(chatPublications).values(rejectedIds.map(id => ({ + companyId: context.fixture.companyId, endpointId: context.endpoint.id, + conversationId: context.conversation.id, issueId: context.conversation.issueId, + idempotencyKey: `run:${id}:working:${context.endpoint.id}`, + payload: { text: "Maya is working…", progressState: "working" }, + state: "published", createdAt: valid.baseCreatedAt, + }))); + const eventAt = new Date(valid.baseCreatedAt.getTime() + 30_000); + await db.insert(heartbeatRunEvents).values(rejectedIds.map(runId => ({ + companyId: source.companyId, runId, agentId: source.agentId, + seq: 1, eventType: "item.completed", + createdAt: sql`${eventAt.toISOString()}::timestamptz + interval '0.000123 seconds'`, + }))); + await context.addEvent(valid, "item.completed", 1, new Date(eventAt.getTime() + 1)); + const select = db.select.bind(db); + let pages = 0; + const guard = vi.spyOn(db, "select").mockImplementation((...args) => { + if (args[0] && "eventCreatedAt" in args[0] && ++pages > 10) { + throw new Error("Native progress pagination did not advance past suppressed candidates"); + } + return select(...args); + }); + try { + await expect(enqueueSafeNativeChatProgress(db, { since: eventAt, limit: 1 })).resolves.toBe(1); + } finally { guard.mockRestore(); } + const progress = await db.select().from(chatPublications).where(and( + eq(chatPublications.conversationId, context.conversation.id), + like(chatPublications.idempotencyKey, "%:native:%"), + )); + expect(progress).toEqual([expect.objectContaining({ + idempotencyKey: `run:${valid.runId}:working:${context.endpoint.id}:native:making_progress:1`, + })]); + } finally { await retirePublicationFixture(context.service, context.endpoint.id); } + }); + it.each(["separate owners", "combined budget"] as const)( "keeps lifecycle and native progress projection independent with %s", async (mode) => { 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__/heartbeat-run-event-atomic-append.test.ts b/server/src/__tests__/heartbeat-run-event-atomic-append.test.ts index dc15a57e6c5..3d4b55b76cb 100644 --- a/server/src/__tests__/heartbeat-run-event-atomic-append.test.ts +++ b/server/src/__tests__/heartbeat-run-event-atomic-append.test.ts @@ -1,16 +1,18 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomBytes } from "node:crypto"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { eq, sql } from "drizzle-orm"; +import { eq, inArray, sql } from "drizzle-orm"; import { drizzle } from "drizzle-orm/postgres-js"; import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ATOMIC_EVENTS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("atomic ordinary run events", () => { - let temporary: Awaited>; + let temporary: Awaited> | undefined; let db: ReturnType; let otherDb: ReturnType; const companyId = "30000000-0000-4000-8000-000000000001"; @@ -30,9 +32,10 @@ describeEmbeddedPostgres("atomic ordinary run events", () => { .where(eq(heartbeatRunEvents.runId, id)).orderBy(heartbeatRunEvents.seq); beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("paperclip-atomic-events-"); - db = createDb(temporary.connectionString); - otherDb = createDb(temporary.connectionString); + if (!externalTestDatabaseUrl) temporary = await startEmbeddedPostgresTestDatabase("paperclip-atomic-events-"); + const connectionString = externalTestDatabaseUrl ?? temporary!.connectionString; + db = createDb(connectionString); + otherDb = createDb(connectionString); await db.insert(companies).values([ { id: companyId, name: "Atomic events", issuePrefix: "ATM" }, { id: otherCompanyId, name: "Other events", issuePrefix: "OTH" }, @@ -42,7 +45,21 @@ describeEmbeddedPostgres("atomic ordinary run events", () => { { id: otherAgentId, companyId: otherCompanyId, name: "Other agent" }, ]); }, 60_000); - afterAll(async () => { await temporary?.cleanup(); }); + afterAll(async () => { + try { + if (externalTestDatabaseUrl && db) { + const fixtureCompanies = [companyId, otherCompanyId]; + await db.delete(heartbeatRunEvents).where(inArray(heartbeatRunEvents.companyId, fixtureCompanies)); + await db.delete(heartbeatRuns).where(inArray(heartbeatRuns.companyId, fixtureCompanies)); + await db.delete(agents).where(inArray(agents.companyId, fixtureCompanies)); + await db.delete(companies).where(inArray(companies.id, fixtureCompanies)); + } + } finally { + await otherDb?.$client.end({ timeout: 1 }); + await db?.$client.end({ timeout: 1 }); + await temporary?.cleanup(); + } + }); it("persists a decoded receipt in one executed statement", async () => { const runId = await makeRun(); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 7ba06dcb5d4..7effe7f6bca 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -270,7 +270,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 96aba422f03..cdba74fb00f 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 74bc9ff9f0c..03514e53dfe 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, publication.conversationId); 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..cc56fd9eac7 --- /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({ companyId: scope.companyId, issueId: scope.issueId })); +} 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 new file mode 100644 index 00000000000..426cedfb1aa --- /dev/null +++ b/server/src/services/chat-native-progress.ts @@ -0,0 +1,427 @@ +import { and, asc, desc, eq, gt, gte, inArray, like, notExists, sql } from "drizzle-orm"; +import { alias } from "drizzle-orm/pg-core"; +import type { Db } from "@paperclipai/db"; +import { agents, chatConversations, chatEndpoints, chatPublications, heartbeatRunEvents, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import { notifyChatPublicationWork } from "./chat-work-notifications.js"; +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, type ChatRunProjectionScope } from "./chat-run-publication-policy.js"; +import { SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, safeNativeChatProgressForEvent } from "./safe-native-chat-progress.js"; + +type SafeNativeChatProgressCandidate = { + eventId: number; + eventSeq: number; + eventType: string; + eventCreatedAt: string; + runId: string; + agentId: string; + issueId: string; + companyId: string; + endpointId: string; + conversationId: string; + agentName: string; +}; + +const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000; + +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", + ); + const laterEvent = alias( + heartbeatRunEvents, + "safe_native_progress_later_event", + ); + const issueIdFromContext = boundRunIssueIdSql(); + return async (pageSize: number, pageCursor: NativeProgressCursor | null): Promise => await db + .select({ + eventId: heartbeatRunEvents.id, + eventSeq: heartbeatRunEvents.seq, + eventType: heartbeatRunEvents.eventType, + // Keep PostgreSQL microseconds in the cursor. Date truncation can + // repeatedly select the same suppressed page without advancing. + eventCreatedAt: sql`${heartbeatRunEvents.createdAt}::text`, + runId: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + issueId: chatConversations.issueId, + companyId: chatConversations.companyId, + endpointId: chatConversations.endpointId, + conversationId: chatConversations.id, + agentName: agents.name, + }) + .from(heartbeatRunEvents) + .innerJoin( + heartbeatRuns, + and( + eq(heartbeatRuns.id, heartbeatRunEvents.runId), + eq(heartbeatRuns.companyId, heartbeatRunEvents.companyId), + eq(heartbeatRuns.agentId, heartbeatRunEvents.agentId), + eq(heartbeatRuns.runtimeMode, "native"), + eq(heartbeatRuns.status, "running"), + ), + ) + .innerJoin( + chatConversations, + and( + eq(chatConversations.companyId, heartbeatRuns.companyId), + sql`coalesce(${heartbeatRuns.issueId}::text, ${issueIdFromContext}) = ${chatConversations.issueId}::text`, + inArray(chatConversations.state, ["active", "waiting"]), + ), + ) + .innerJoin( + chatEndpoints, + and( + eq(chatEndpoints.companyId, chatConversations.companyId), + eq(chatEndpoints.id, chatConversations.endpointId), + eq(chatEndpoints.publicationMode, "automatic"), + allowsAutomaticRunPublication(), + eq(chatEndpoints.assignedAgentId, heartbeatRuns.agentId), + ), + ) + .innerJoin(agents, eq(agents.id, heartbeatRuns.agentId)) + .innerJoin( + basePublication, + and( + eq(basePublication.companyId, heartbeatRuns.companyId), + eq(basePublication.endpointId, chatConversations.endpointId), + eq(basePublication.conversationId, chatConversations.id), + eq(basePublication.issueId, chatConversations.issueId), + sql`${basePublication.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${chatConversations.endpointId}::text`, + inArray(basePublication.state, [ + "pending", + "retry", + "streaming", + "published", + ]), + ), + ) + .where( + and( + inArray(heartbeatRunEvents.eventType, [ + ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, + ]), + 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 + .select({ id: laterEvent.id }) + .from(laterEvent) + .where( + and( + eq(laterEvent.companyId, heartbeatRunEvents.companyId), + eq(laterEvent.runId, heartbeatRunEvents.runId), + gt(laterEvent.seq, heartbeatRunEvents.seq), + inArray(laterEvent.eventType, [ + ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, + ]), + ), + ), + ), + pageCursor + ? sql`(${heartbeatRunEvents.createdAt}, ${heartbeatRuns.id}, ${chatConversations.id}) > + (${pageCursor.eventCreatedAt}::timestamptz, ${pageCursor.runId}::uuid, ${pageCursor.conversationId}::uuid)` + : undefined, + ), + ) + .orderBy( + asc(heartbeatRunEvents.createdAt), + asc(heartbeatRuns.id), + 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 = { + eventCreatedAt: lastRow.eventCreatedAt, + runId: lastRow.runId, + conversationId: lastRow.conversationId, + }; + + 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 + // publication insert takes an issue FK lock, so locking the run first + // would invert that order against a concurrent terminal commit. A + // contended issue/run is optional progress, not a reason to hold other + // chats: skip it now and revisit it with a fresh cursor next sweep. + const currentIssue = await tx + .select({ id: issues.id }) + .from(issues) + .where( + and( + eq(issues.id, row.issueId), + eq(issues.companyId, row.companyId), + ), + ) + .for("update", { skipLocked: true }) + .then((currentRows) => currentRows[0] ?? null); + 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. + const currentRun = 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"), + ), + ) + .for("update", { skipLocked: true }) + .then((currentRows) => currentRows[0] ?? null); + 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 }) + .from(chatConversations) + .innerJoin( + chatEndpoints, + and( + eq(chatEndpoints.companyId, chatConversations.companyId), + eq(chatEndpoints.id, chatConversations.endpointId), + eq(chatEndpoints.publicationMode, "automatic"), + allowsAutomaticRunPublication(), + eq(chatEndpoints.assignedAgentId, row.agentId), + ), + ) + .where( + and( + eq(chatConversations.companyId, row.companyId), + eq(chatConversations.id, row.conversationId), + eq(chatConversations.endpointId, row.endpointId), + eq(chatConversations.issueId, row.issueId), + inArray(chatConversations.state, ["active", "waiting"]), + ), + ) + .limit(1) + .then((currentRows) => currentRows[0] ?? null); + if (!destination) return 0; + + const [currentEvent] = await tx + .select({ + id: heartbeatRunEvents.id, + seq: heartbeatRunEvents.seq, + eventType: heartbeatRunEvents.eventType, + createdAt: heartbeatRunEvents.createdAt, + }) + .from(heartbeatRunEvents) + .where( + and( + eq(heartbeatRunEvents.companyId, row.companyId), + eq(heartbeatRunEvents.runId, row.runId), + inArray(heartbeatRunEvents.eventType, [ + ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, + ]), + ), + ) + .orderBy(desc(heartbeatRunEvents.seq), desc(heartbeatRunEvents.id)) + .limit(1); + if ( + !currentEvent || + currentEvent.id !== row.eventId || + currentEvent.seq !== row.eventSeq || + currentEvent.eventType !== row.eventType + ) { + return 0; + } + const progress = safeNativeChatProgressForEvent( + currentEvent.eventType, + row.agentName, + ); + if (!progress) return 0; + + const baseKey = `run:${row.runId}:working:${row.endpointId}`; + const [currentBase] = await tx + .select({ createdAt: chatPublications.createdAt }) + .from(chatPublications) + .where( + and( + eq(chatPublications.companyId, row.companyId), + eq(chatPublications.endpointId, row.endpointId), + eq(chatPublications.conversationId, row.conversationId), + eq(chatPublications.issueId, row.issueId), + eq(chatPublications.idempotencyKey, baseKey), + inArray(chatPublications.state, [ + "pending", + "retry", + "streaming", + "published", + ]), + ), + ) + .limit(1); + if (!currentBase) return 0; + + const progressKeyPrefix = `${baseKey}:native:`; + const previousProgress = await tx + .select({ + createdAt: chatPublications.createdAt, + idempotencyKey: chatPublications.idempotencyKey, + }) + .from(chatPublications) + .where( + and( + eq(chatPublications.companyId, row.companyId), + eq(chatPublications.endpointId, row.endpointId), + eq(chatPublications.conversationId, row.conversationId), + eq(chatPublications.issueId, row.issueId), + like(chatPublications.idempotencyKey, `${progressKeyPrefix}%`), + ), + ); + if ( + previousProgress.some((publication) => + publication.idempotencyKey.startsWith( + `${progressKeyPrefix}${progress.phase}:`, + ), + ) + ) { + return 0; + } + const lastProgressAt = Math.max( + currentBase.createdAt.getTime(), + ...previousProgress.map((publication) => + publication.createdAt.getTime(), + ), + ); + if ( + currentEvent.createdAt.getTime() - lastProgressAt < + SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS + ) { + return 0; + } + + const bindings = await resolveChatOriginPublicationBindings( + tx, + row.companyId, + row.issueId, + row.runId, + ); + if ( + !bindings.some( + (binding) => + binding.endpointId === row.endpointId && + binding.conversationId === row.conversationId, + ) + ) { + return 0; + } + if ( + await hasChatRunOwnedProviderInteraction(tx, { + companyId: row.companyId, + issueId: row.issueId, + runId: row.runId, + }) + ) { + return 0; + } + const explicitlyAuthoredFinal = await tx + .select({ id: chatPublications.id }) + .from(chatPublications) + .innerJoin( + issueComments, + and( + eq(issueComments.companyId, chatPublications.companyId), + eq(issueComments.id, chatPublications.commentId), + ), + ) + .where( + and( + eq(chatPublications.companyId, row.companyId), + eq(chatPublications.endpointId, row.endpointId), + eq(chatPublications.conversationId, row.conversationId), + eq(chatPublications.issueId, row.issueId), + eq(issueComments.authorType, "agent"), + eq(issueComments.createdByRunId, row.runId), + sql`( + ${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol' + or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_' + )`, + ), + ) + .limit(1); + if (explicitlyAuthoredFinal.length > 0) return 0; + + await notifyChatPublicationWork(tx); + const insertedRows = await tx + .insert(chatPublications) + .values({ + companyId: row.companyId, + endpointId: row.endpointId, + conversationId: row.conversationId, + issueId: row.issueId, + idempotencyKey: `${progressKeyPrefix}${progress.phase}:${currentEvent.seq}`, + payload: projectSafeChatPublication({ + classification: "external", + source: "safe_milestone", + text: progress.text, + progressState: "working", + }), + state: "pending", + }) + .onConflictDoNothing() + .returning({ id: chatPublications.id }); + return insertedRows.length; + }); + inserted += result; + } + if (rows.length < pageSize) break; + } + 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 new file mode 100644 index 00000000000..0ce57ff5794 --- /dev/null +++ b/server/src/services/chat-run-publication-policy.ts @@ -0,0 +1,21 @@ +import { ne, notExists, or, sql } from "drizzle-orm"; +import { chatEndpoints, chatGitHubConfigurations } from "@paperclipai/db"; + +// Only upgraded GitHub bots have the task-scoped response tools. Legacy +// connections retain automatic progress and final-answer publication. +export function allowsAutomaticRunPublication() { + return or(ne(chatEndpoints.provider, "github"), notExists( + sql`(select 1 from ${chatGitHubConfigurations} + where ${chatGitHubConfigurations.companyId} = ${chatEndpoints.companyId} + and ${chatGitHubConfigurations.endpointId} = ${chatEndpoints.id})`, + )); +} + +export function chatRunProjectionWindow(input: { since?: Date; limit?: number }) { + return { + since: input.since ?? new Date(Date.now() - 24 * 60 * 60_000), + 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 e6a1ac317a4..84ea1783dd7 100644 --- a/server/src/services/chat-run-publications.ts +++ b/server/src/services/chat-run-publications.ts @@ -1,21 +1,22 @@ +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"; import type { ChatRunMilestoneScope } from "./chat-run-milestone-notifications.js"; import { notifyChatPublicationWork } from "./chat-work-notifications.js"; import { and, asc, - desc, eq, gt, gte, inArray, isNull, - like, ne, notExists, or, sql, } from "drizzle-orm"; -import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; import { agents, @@ -26,7 +27,6 @@ import { chatGitHubReviews, chatMessageLinks, chatPublications, - heartbeatRunEvents, heartbeatRuns, issueComments, issues, @@ -39,10 +39,6 @@ import { resolveChatOriginPublicationBindings } from "./issues.js"; import { boundRunIssueIdSql } from "./run-task-binding.js"; import { chatRunMilestoneSource } from "./chat-run-milestone-candidates.js"; import { authorizeNativeChatReviewPresentation } from "./native-runtime/native-chat-review-presentation.js"; -import { - SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, - safeNativeChatProgressForEvent, -} from "./safe-native-chat-progress.js"; type SafeRunMilestone = "queued" | "working" | "waiting_for_input" | "completed" | "failed"; @@ -54,15 +50,6 @@ const OWNERSHIP_ATTENTION_CODES = [ export { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON }; -// Only upgraded GitHub bots have the task-scoped response tools. Legacy -// connections retain automatic progress and final-answer publication. -function allowsAutomaticRunPublication() { - return or(ne(chatEndpoints.provider, "github"), notExists( - sql`(select 1 from ${chatGitHubConfigurations} - where ${chatGitHubConfigurations.companyId} = ${chatEndpoints.companyId} - and ${chatGitHubConfigurations.endpointId} = ${chatEndpoints.id})`, - )); -} /** * Heartbeat's presentation resolver may externalize its selected final prose @@ -227,21 +214,6 @@ type ChatRunMilestoneCandidate = { agentName: string; }; -type SafeNativeChatProgressCandidate = { - eventId: number; - eventSeq: number; - eventType: string; - eventCreatedAt: Date; - runId: string; - agentId: string; - issueId: string; - companyId: string; - endpointId: string; - conversationId: string; - agentName: string; -}; - -const SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS = 20_000; function milestoneForStatus( status: string, @@ -293,384 +265,6 @@ export function safeMilestoneText(input: { }`; } -/** - * Projects a bounded sample of native activity into the existing run working - * lane. The selector intentionally reads only event identity, type, sequence, - * and time; native messages and payloads stay inside Paperclip. - */ -export async function enqueueSafeNativeChatProgress( - db: Db, - options: { since?: Date; limit?: number } = {}, -): Promise { - const input = chatRunProjectionWindow(options); - const basePublication = alias( - chatPublications, - "safe_native_progress_base_publication", - ); - const laterEvent = alias( - heartbeatRunEvents, - "safe_native_progress_later_event", - ); - const issueIdFromContext = boundRunIssueIdSql(); - let inserted = 0; - let cursor: { - eventCreatedAt: Date; - 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 - .select({ - eventId: heartbeatRunEvents.id, - eventSeq: heartbeatRunEvents.seq, - eventType: heartbeatRunEvents.eventType, - eventCreatedAt: heartbeatRunEvents.createdAt, - runId: heartbeatRuns.id, - agentId: heartbeatRuns.agentId, - issueId: chatConversations.issueId, - companyId: chatConversations.companyId, - endpointId: chatConversations.endpointId, - conversationId: chatConversations.id, - agentName: agents.name, - }) - .from(heartbeatRunEvents) - .innerJoin( - heartbeatRuns, - and( - eq(heartbeatRuns.id, heartbeatRunEvents.runId), - eq(heartbeatRuns.companyId, heartbeatRunEvents.companyId), - eq(heartbeatRuns.agentId, heartbeatRunEvents.agentId), - eq(heartbeatRuns.runtimeMode, "native"), - eq(heartbeatRuns.status, "running"), - ), - ) - .innerJoin( - chatConversations, - and( - eq(chatConversations.companyId, heartbeatRuns.companyId), - sql`coalesce(${heartbeatRuns.issueId}::text, ${issueIdFromContext}) = ${chatConversations.issueId}::text`, - inArray(chatConversations.state, ["active", "waiting"]), - ), - ) - .innerJoin( - chatEndpoints, - and( - eq(chatEndpoints.companyId, chatConversations.companyId), - eq(chatEndpoints.id, chatConversations.endpointId), - eq(chatEndpoints.publicationMode, "automatic"), - allowsAutomaticRunPublication(), - eq(chatEndpoints.assignedAgentId, heartbeatRuns.agentId), - ), - ) - .innerJoin(agents, eq(agents.id, heartbeatRuns.agentId)) - .innerJoin( - basePublication, - and( - eq(basePublication.companyId, heartbeatRuns.companyId), - eq(basePublication.endpointId, chatConversations.endpointId), - eq(basePublication.conversationId, chatConversations.id), - eq(basePublication.issueId, chatConversations.issueId), - sql`${basePublication.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${chatConversations.endpointId}::text`, - inArray(basePublication.state, [ - "pending", - "retry", - "streaming", - "published", - ]), - ), - ) - .where( - and( - inArray(heartbeatRunEvents.eventType, [ - ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, - ]), - gte(heartbeatRunEvents.createdAt, input.since), - sql`${heartbeatRunEvents.createdAt} >= ${basePublication.createdAt} + interval '20 seconds'`, - notExists( - db - .select({ id: laterEvent.id }) - .from(laterEvent) - .where( - and( - eq(laterEvent.companyId, heartbeatRunEvents.companyId), - eq(laterEvent.runId, heartbeatRunEvents.runId), - gt(laterEvent.seq, heartbeatRunEvents.seq), - inArray(laterEvent.eventType, [ - ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, - ]), - ), - ), - ), - pageCursor - ? or( - gt(heartbeatRunEvents.createdAt, pageCursor.eventCreatedAt), - and( - eq(heartbeatRunEvents.createdAt, pageCursor.eventCreatedAt), - or( - gt(heartbeatRuns.id, pageCursor.runId), - and( - eq(heartbeatRuns.id, pageCursor.runId), - gt(chatConversations.id, pageCursor.conversationId), - ), - ), - ), - ) - : undefined, - ), - ) - .orderBy( - asc(heartbeatRunEvents.createdAt), - asc(heartbeatRuns.id), - asc(chatConversations.id), - ) - .limit(pageSize); - if (rows.length === 0) break; - const lastRow = rows.at(-1)!; - cursor = { - eventCreatedAt: lastRow.eventCreatedAt, - runId: lastRow.runId, - conversationId: lastRow.conversationId, - }; - - for (const row of rows) { - if (inserted >= input.limit) break; - const result = await db.transaction(async (tx) => { - // Keep the normal issue -> run order used by native finalization. The - // publication insert takes an issue FK lock, so locking the run first - // would invert that order against a concurrent terminal commit. A - // contended issue/run is optional progress, not a reason to hold other - // chats: skip it now and revisit it with a fresh cursor next sweep. - const currentIssue = await tx - .select({ id: issues.id }) - .from(issues) - .where( - and( - eq(issues.id, row.issueId), - eq(issues.companyId, row.companyId), - ), - ) - .for("update", { skipLocked: true }) - .then((currentRows) => currentRows[0] ?? null); - if (!currentIssue) 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. - const currentRun = 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"), - ), - ) - .for("update", { skipLocked: true }) - .then((currentRows) => currentRows[0] ?? null); - if (!currentRun) return 0; - - const destination = await tx - .select({ id: chatConversations.id }) - .from(chatConversations) - .innerJoin( - chatEndpoints, - and( - eq(chatEndpoints.companyId, chatConversations.companyId), - eq(chatEndpoints.id, chatConversations.endpointId), - eq(chatEndpoints.publicationMode, "automatic"), - allowsAutomaticRunPublication(), - eq(chatEndpoints.assignedAgentId, row.agentId), - ), - ) - .where( - and( - eq(chatConversations.companyId, row.companyId), - eq(chatConversations.id, row.conversationId), - eq(chatConversations.endpointId, row.endpointId), - eq(chatConversations.issueId, row.issueId), - inArray(chatConversations.state, ["active", "waiting"]), - ), - ) - .limit(1) - .then((currentRows) => currentRows[0] ?? null); - if (!destination) return 0; - - const [currentEvent] = await tx - .select({ - id: heartbeatRunEvents.id, - seq: heartbeatRunEvents.seq, - eventType: heartbeatRunEvents.eventType, - createdAt: heartbeatRunEvents.createdAt, - }) - .from(heartbeatRunEvents) - .where( - and( - eq(heartbeatRunEvents.companyId, row.companyId), - eq(heartbeatRunEvents.runId, row.runId), - inArray(heartbeatRunEvents.eventType, [ - ...SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES, - ]), - ), - ) - .orderBy(desc(heartbeatRunEvents.seq), desc(heartbeatRunEvents.id)) - .limit(1); - if ( - !currentEvent || - currentEvent.id !== row.eventId || - currentEvent.seq !== row.eventSeq || - currentEvent.eventType !== row.eventType - ) { - return 0; - } - const progress = safeNativeChatProgressForEvent( - currentEvent.eventType, - row.agentName, - ); - if (!progress) return 0; - - const baseKey = `run:${row.runId}:working:${row.endpointId}`; - const [currentBase] = await tx - .select({ createdAt: chatPublications.createdAt }) - .from(chatPublications) - .where( - and( - eq(chatPublications.companyId, row.companyId), - eq(chatPublications.endpointId, row.endpointId), - eq(chatPublications.conversationId, row.conversationId), - eq(chatPublications.issueId, row.issueId), - eq(chatPublications.idempotencyKey, baseKey), - inArray(chatPublications.state, [ - "pending", - "retry", - "streaming", - "published", - ]), - ), - ) - .limit(1); - if (!currentBase) return 0; - - const progressKeyPrefix = `${baseKey}:native:`; - const previousProgress = await tx - .select({ - createdAt: chatPublications.createdAt, - idempotencyKey: chatPublications.idempotencyKey, - }) - .from(chatPublications) - .where( - and( - eq(chatPublications.companyId, row.companyId), - eq(chatPublications.endpointId, row.endpointId), - eq(chatPublications.conversationId, row.conversationId), - eq(chatPublications.issueId, row.issueId), - like(chatPublications.idempotencyKey, `${progressKeyPrefix}%`), - ), - ); - if ( - previousProgress.some((publication) => - publication.idempotencyKey.startsWith( - `${progressKeyPrefix}${progress.phase}:`, - ), - ) - ) { - return 0; - } - const lastProgressAt = Math.max( - currentBase.createdAt.getTime(), - ...previousProgress.map((publication) => - publication.createdAt.getTime(), - ), - ); - if ( - currentEvent.createdAt.getTime() - lastProgressAt < - SAFE_NATIVE_CHAT_PROGRESS_CADENCE_MS - ) { - return 0; - } - - const bindings = await resolveChatOriginPublicationBindings( - tx, - row.companyId, - row.issueId, - row.runId, - ); - if ( - !bindings.some( - (binding) => - binding.endpointId === row.endpointId && - binding.conversationId === row.conversationId, - ) - ) { - return 0; - } - if ( - await hasChatRunOwnedProviderInteraction(tx, { - companyId: row.companyId, - issueId: row.issueId, - runId: row.runId, - }) - ) { - return 0; - } - const explicitlyAuthoredFinal = await tx - .select({ id: chatPublications.id }) - .from(chatPublications) - .innerJoin( - issueComments, - and( - eq(issueComments.companyId, chatPublications.companyId), - eq(issueComments.id, chatPublications.commentId), - ), - ) - .where( - and( - eq(chatPublications.companyId, row.companyId), - eq(chatPublications.endpointId, row.endpointId), - eq(chatPublications.conversationId, row.conversationId), - eq(chatPublications.issueId, row.issueId), - eq(issueComments.authorType, "agent"), - eq(issueComments.createdByRunId, row.runId), - sql`( - ${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol' - or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_' - )`, - ), - ) - .limit(1); - if (explicitlyAuthoredFinal.length > 0) return 0; - - await notifyChatPublicationWork(tx); - const insertedRows = await tx - .insert(chatPublications) - .values({ - companyId: row.companyId, - endpointId: row.endpointId, - conversationId: row.conversationId, - issueId: row.issueId, - idempotencyKey: `${progressKeyPrefix}${progress.phase}:${currentEvent.seq}`, - payload: projectSafeChatPublication({ - classification: "external", - source: "safe_milestone", - text: progress.text, - progressState: "working", - }), - state: "pending", - }) - .onConflictDoNothing() - .returning({ id: chatPublications.id }); - return insertedRows.length; - }); - inserted += result; - } - if (rows.length < pageSize) break; - } - return inserted; -} type ChatRunMilestoneCursor = { updatedAt: string; @@ -1096,6 +690,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) @@ -1135,12 +732,6 @@ type ChatRunProjectionInput = { limit?: number; }; -function chatRunProjectionWindow(input: ChatRunProjectionInput) { - return { - since: input.since ?? new Date(Date.now() - 24 * 60 * 60_000), - limit: Math.max(1, Math.min(input.limit ?? 200, 1_000)), - }; -} /** Explicit combined reconciliation retains lifecycle priority and one shared * publication budget. Background owners can reconcile the two duties separately. */ diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index e4434d9136f..45f9e520f8d 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -57,6 +57,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 efefbc63b3c..53e0bd26206 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,10 @@ 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, + // Only progress needs the task binding. Other native/recovery events + // must not read a potentially large context snapshot to append a log. + ...(projectsProgress ? { runtimeMode: heartbeatRuns.runtimeMode, issueId: boundRunIssueIdSql() } : {}) }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .for("update") @@ -165,6 +172,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)