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 2f6c23a2c4c..0db605ede5c 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2327,3 +2327,15 @@ full local/remote acceptance, and PR publication are still required. Master was refreshed to `7703220d38`; the complete stack has not yet been reconciled with it. No merge, complete migration, whole-instance sleep, or full-goal completion is claimed. + +## Slice 48 producer and lifecycle checkpoint + +Response recovery now shares the finalization notification boundary, receives +run-scoped late evidence, and wakes when a paused agent resumes or retry admission +settles. The resume gap was reproduced before fixing it at lifecycle's transaction +boundary. The existing lifecycle suite, both response variants' supporting suites, +and the combined acceptance selection cover 175 distinct passing cases across +runs; server types pass. A real source-row lock verifies retry across drain. +Earlier harness/test failures and the 10,000-unpresented-row discovery measurements +are preserved in the response audit. No startup wiring, periodic removal, PR +publication, remote acceptance, or whole-instance sleep is claimed yet. diff --git a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md index 5106a4ba796..4a2f067ae37 100644 --- a/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md +++ b/doc/plans/2026-10-25-native-response-presentation-sleep-audit.md @@ -230,3 +230,84 @@ cases in `/tmp/sleep48-worker-tests-v4.log`; the final expanded result is No slice 48 PR, full workspace acceptance, remote CI/review, or provider sleep qualification is claimed by these local results. The earlier stack gates and remaining recurring-server duties are unchanged. + +## Producer integration and lifecycle evidence + +The existing pending-effect notification boundary is now named +`notifyNativeFinalizationWork` and wakes both consumers. Its existing coordinator +claim/retry, effect settlement, persisted-attention replay, retained-session release, +and recovery callers retain transaction ownership. No new wrappers are threaded +through those callers. Response discovery now accepts exact run and agent scopes. + +Both journal writers notify for accepted-result evidence and normalized completed +final assistant messages, after source/deduplication checks and before persistence. +Deltas, tool completions, and analysis messages do not activate response recovery. +The port's late accepted-result journal test and the internal final-message test +exercise these existing writers. Monitor continuation recovery is also exercised +through the worker, without another accepted result or provider execution. + +Agent availability was a real missing producer: the recovery owner rejects a +paused agent, but resuming that agent did not invalidate dormant responses. A +real resume regression failed before the fix. Lifecycle now exposes a narrow +`beforeStatusChange` integration callback, invoked inside its existing transaction +before the compatibility status write. The service supplies response policy only +for a transition from non-invokable to invokable, scoped to that agent. Lifecycle +still owns status transitions and the transaction; its existing resource-event +callback retains its semantics. With no response worker registered, this adds only +the existing notification protocol, not an independent worker or provider call. +The complete lifecycle suite exercises existing consumers of the new contract. + +Failed-chat retry admission also notifies response recovery when its source action +becomes processed. This can establish the ancestry proof after a very fast child +run has already finished. Five real admission scenarios verify that the response +consumer observes the committed action and the exact issue scope: commit, retry, +startup, stale claim, and rollback. An assignment-reactivation experiment instead +hit the existing prohibition on changing a channel-bound task's agent; that +unsupported fixture was removed without relaxing the guard. + +Validation across these runs covers 175 distinct cases: 59 response/authority and +retry-admission cases, 58 lifecycle cases, 21 monitor-continuation cases, and 37 +control-plane-port cases. Server typechecking passes. These are selected suites, +not a claim that the entire repository suite or remote PR checks passed. + +- `/tmp/sleep48-producers-rerun.log`: port 37/37, continuation 20/20, server types. +- `/tmp/sleep48-producer-acceptance.log`: lifecycle 58/58, expanded continuation + 21/21, response/authority 53 selected, server types. +- `/tmp/sleep48-acceptance-v5.log`: combined response/authority/admission 59 passed; + 1,546 unrelated cases not selected. Types in + `/tmp/sleep48-acceptance-typecheck-v5.log`. +- Real action-row contention now verifies retry retention through drain, zero + passive SQL while paused, and eventual single publication after resume. + +Failures remain recorded. The first external-port run had 36 passes and one +observer connection using the absent embedded database; the observer now uses the +validated external URL. The resume regression and invalid reassignment experiment +are in `/tmp/sleep48-eligibility-before.log`. A test-only observer initially omitted +its required retry interval and produced finite-deadline warnings; corrected in +`/tmp/sleep48-retry-producer-v2.log`. The combined v4 run exposed a global fault +injection targeting another retained fixture (13 transactions, not one). The v5 +run uses an actual lock on the tested response's source instead of assuming global +candidate order; all 59 cases pass with that stronger fault boundary. + +## Unpresented-backlog measurement + +A disposable clone of the earlier plan fixture has 10,008 structurally eligible, +unpresented candidates. The actual worker loader took 2,980 ms and 101 queries for +global reconstruction; 10,001 candidates in one company took 2,828 ms; the exact +run took 1.73 ms and one query. Agent reconstruction for that company's 10,001 +candidates took 2,826 ms. Serialized entries occupy about 2.05 MB. The drain gate +stopped discovery after two pages. Evidence: +`/tmp/sleep48-unpresented-backlog.mts` and `.log`. + +This measures discovery, not authorization or publication throughput: the cloned +rows do not establish valid independent publication authority. Each SQL page is +bounded to 100 rows, but total reconstruction memory/time still scales with the +unpresented backlog. Ordinary finalization and journal notifications are run-scoped; +agent resume is agent-scoped; actual authority/configuration changes may reconstruct +a larger scope. Health-only configuration signals do not invalidate this worker. +Do not describe total reconstruction as constant-cost or claim fleet qualification. + +Startup/shutdown integration, removal of this duty's periodic call, final producer +coverage review, full PR verification, and publication remain outstanding. The old +backstop is still present. Master was refreshed to `6bdcb70244`; the full stack +still needs reconciliation and its earlier failing/review gates remain open. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 32bc05faddb..46c545bd45b 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -15,7 +15,7 @@ import { accessService } from "../services/access.js"; import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; import { restoreNativeChatReviewPresentationInTransaction } from "../services/native-runtime/native-chat-review-presentation.js"; import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; -import { configureAgentLifecycle } from "../services/agent-lifecycle.js"; +import { configureAgentLifecycle, createAgentLifecycle, processAgentLifecycle } from "../services/agent-lifecycle.js"; import { deleteCompany } from "../services/company-deletion.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js"; @@ -63993,6 +63993,19 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const input = { companyId: context.fixture.companyId, issueId: context.issue.id, agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user" }; const stage = () => db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, input)); + const observedResponseAdmissions: string[] = []; + const responseHints: string[] = []; + await coordinator.register(DELIVERY_QUEUES.nativeResponsePresentation, { + retryMs: 30_000, + onSettled: hints => { responseHints.push(...hints ?? []); }, + run: async () => { + const admitted = await db.select({ id: chatActions.id }).from(chatActions).where(and( + eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "failed_run_retry"), eq(chatActions.status, "processed"), + )); + observedResponseAdmissions.push(...admitted.map(row => row.id)); + }, + hasPending: async () => false, + }).ready; context.wakeup.mockClear(); try { let actionId: string; @@ -64038,6 +64051,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(action?.result.attemptCount).toBe(mode === "retry" ? 2 : 1); expect(coordinator.nextWakeAt()).toBeNull(); }, { timeout: 10_000 }); + await vi.waitFor(() => expect(observedResponseAdmissions).toContain(actionId), { timeout: 5000 }); + expect(responseHints.map(hint => JSON.parse(hint))).toContainEqual({ companyId: context.fixture.companyId, issueId: context.issue.id }); expect(context.wakeup).toHaveBeenCalledTimes(mode === "retry" ? 2 : 1); const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, actionId)); expect(receipts).toHaveLength(1); @@ -72702,6 +72717,49 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } }); + it("retains a failed accepted-response attempt across drain and retries after resume", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + let release!: () => void, observed!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { observed = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(chatActions).where(eq(chatActions.id, context.action.id)).for("update"); + observed(); + await hold; + }); + await locked; + let canRun = true; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => canRun); + try { + await worker.ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(errors).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + canRun = false; + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + release(); await holder; + canRun = true; worker.wake(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + expect(errors).toHaveLength(1); + } finally { release(); await holder; worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + it("reconstructs a future accepted-response deadline without early publication", async () => { const context = await committedChatResponseRecoveryFixture("telegram"); const at = new Date(Date.now() + 60_000); @@ -72750,6 +72808,88 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { intercept.mockRestore(); vi.useRealTimers(); await context.service.shutdown(); } }); + it("wakes dormant accepted-response recovery after agent resume", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + await db.update(agents).set({ status: "paused", lifecycleState: "paused", lifecycleHolds: ["manual"], lifecycleOperation: null }) + .where(eq(agents.id, context.binding.agentId)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const work = nativeResponsePresentationWork(db); + const worker = work.register(coordinator, () => true); + const lifecycle = configureAgentLifecycle(db, { + requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete", + }); + try { + await worker.ready; + expect(coordinator.nextWakeAt()).toBeNull(); + expect(await work.load({ companyId: context.fixture.companyId, agentId: randomUUID() })).toEqual([]); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await createAgentLifecycle(db).resumeAgent(context.binding.agentId); + await processAgentLifecycle(db, context.binding.agentId); + await expect.poll(async () => (await db.select().from(agents).where(eq(agents.id, context.binding.agentId)))[0].status).toBe("idle"); + await expect.poll(async () => (await readChatPublicationsByEndpointId(context.endpoint.id)).length, { timeout: 5000 }).toBe(1); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await lifecycle.stop(); worker.stop(); await coordinator.stop(); await context.service.shutdown(); } + }); + + it("wakes accepted-response recovery from the real coordinator retry release", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const store = new NativeRunCoordinatorStore(db, context.binding); + await store.claimFinalization({ leaseOwner: "response-recovery-fixture", leaseTtlMs: 1000 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + try { + await worker.ready; + expect(coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(2000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + expect(coordinator.nextWakeAt()).toBeNull(); + await store.markFinalizationRetry({ leaseOwner: "response-recovery-fixture", failureCode: "response_fixture_retry", retryAfterMs: 1000 }); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("wakes accepted-response recovery when its authenticated accepted-result journal arrives late", async () => { + const context = await committedChatResponseRecoveryFixture("telegram", "port"); + await db.delete(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, context.runId), eq(heartbeatRunEvents.eventType, "run.result.accepted"))); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + try { + await worker.ready; + expect(coordinator.nextWakeAt()).toBeNull(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + const port = new PaperclipControlPlanePort(db, { + companyId: context.binding.companyId, issueId: context.binding.issueId, + runId: context.runId, agentId: context.binding.agentId, sessionId: context.binding.normalizedSessionId, + completionContractId: context.binding.completionContractId, + completionContractSha256: context.binding.completionContractSha256, + sourceInstanceId: context.binding.runnerSourceInstanceId, controlPlaneSourceInstanceId: `control-${context.runId}`, + }); + await port.appendEvent({ + schema: "paperclip.prp.event.v1", sourceEventId: `accepted-${context.runId}`, sourceSeq: 1, + sourceInstanceId: `control-${context.runId}`, sourceKind: "control_plane", + runId: context.runId, normalizedSessionId: context.binding.normalizedSessionId, + turnId: `turn-${context.runId}`, eventType: "run.result.accepted", schemaVersion: 1, + priority: 0, emittedAt: new Date().toISOString(), payload: { result: context.result }, + }); + await expect.poll(async () => ({ + publications: (await readChatPublicationsByEndpointId(context.endpoint.id)).length, + active: idleWorkSnapshot().active, next: coordinator.nextWakeAt(), + }), { timeout: 5000 }).toEqual({ publications: 1, active: 0, next: null }); + expect(errors).toEqual([]); + } finally { worker.stop(); await coordinator.stop(); await context.service.shutdown(); } + }); + it("defers scheduled accepted-response recovery until the persisted retry is due", async () => { const context = await committedChatResponseRecoveryFixture("telegram"); const input = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId, onlyWhenDue: true }; diff --git a/server/src/__tests__/committed-chat-response-authority.test.ts b/server/src/__tests__/committed-chat-response-authority.test.ts index 497cbaa4233..a09b8847c1b 100644 --- a/server/src/__tests__/committed-chat-response-authority.test.ts +++ b/server/src/__tests__/committed-chat-response-authority.test.ts @@ -1,3 +1,4 @@ +import { isNativeResponseEvidence } from "../services/native-response-presentation-notifications.js"; import { describe, expect, it, vi } from "vitest"; import type { Db } from "@paperclipai/db"; import { authorizeCommittedChatResponse, onCommittedChatResponseAuthorityRegistered, registerCommittedChatResponseAuthority } from "../services/durable-chat-wakeup.js"; @@ -39,3 +40,16 @@ describe("committed response authority lifecycle", () => { } finally { removeSecond(); } }); }); + + +describe("accepted response evidence selection", () => { + it.each([ + { type: "item.delta", kind: "agentMessage", channel: "final", expected: false }, + { type: "item.completed", kind: "agentMessage", channel: "analysis", expected: false }, + { type: "item.completed", kind: "commandExecution", channel: "final", expected: false }, + { type: "item.completed", kind: "agentMessage", channel: "final", expected: true }, + { type: "run.result.accepted", kind: "agentMessage", channel: "final", expected: true }, + ])("signals only response evidence: $type/$kind/$channel", ({ type, kind, channel, expected }) => { + expect(isNativeResponseEvidence(type, { payload: { kind, channel, text: "Saved evidence" } })).toBe(expected); + }); +}); diff --git a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts index 681a6ec6735..2f8e0d0724e 100644 --- a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; import { createAgentLifecycle as createLifecycleCommands } from "../index.js"; import { createAgentLifecycleEffects } from "../../../services/agent-lifecycle.js"; import { deleteCompany } from "../../../services/company-deletion.js"; @@ -28,13 +29,20 @@ import { budgetService } from "../../../services/budgets.js"; const createLifecycleStore = (db: Db) => createStore(db, createAgentLifecycleEffects()); -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_LIFECYCLE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("agent lifecycle commands", () => { - let database: Awaited>; + let database: Awaited> | undefined; let db: Db; let companyId: string; const workers: ReturnType[] = []; - beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("paperclip-agent-lifecycle-"); db = createDb(database.connectionString); }, 90_000); + beforeAll(async () => { + if (external) db = createDb(external); + else { + database = await startEmbeddedPostgresTestDatabase("paperclip-agent-lifecycle-"); + db = createDb(database.connectionString); + } + }, 90_000); afterAll(async () => { for (const worker of workers) await worker.stop(); await database?.cleanup(); }); beforeEach(async () => { for (const work of workers) await work.stop(); diff --git a/server/src/modules/agent-lifecycle/adapters/effects.ts b/server/src/modules/agent-lifecycle/adapters/effects.ts index 4c12d36866e..b3716f9fc02 100644 --- a/server/src/modules/agent-lifecycle/adapters/effects.ts +++ b/server/src/modules/agent-lifecycle/adapters/effects.ts @@ -7,6 +7,7 @@ import type { ActivityPublication } from "../../../types/activity-publication.js /** Services supply integrations; lifecycle owns the transaction and state writes. */ export interface LifecycleEffects { notifyWork(db: Db): Promise; + beforeStatusChange(db: Db, companyId: string, agentId: string, before: string, after: string): Promise; reportFailure(context: LifecycleFailure, error: unknown): void; deleteDependencies(db: Db, companyId: string, agentId: string): Promise; prepareConfiguration(db: Db, current: AgentConfigurationRecord, data: AgentConfigurationPatch, options?: UpdateAgentOptions): Promise; diff --git a/server/src/modules/agent-lifecycle/adapters/postgres.ts b/server/src/modules/agent-lifecycle/adapters/postgres.ts index 2d885a4f1c1..a77aed0d7c7 100644 --- a/server/src/modules/agent-lifecycle/adapters/postgres.ts +++ b/server/src/modules/agent-lifecycle/adapters/postgres.ts @@ -71,6 +71,7 @@ async function changeLockedAgent(tx: Db, effects: LifecycleEffects, agent: Lifec }; if (next === "ready") operation.resumeState = "ready"; const status = compatibilityStatus(next, agent.status); + if (status !== agent.status) await effects.beforeStatusChange(tx, agent.companyId, agent.id, agent.status, status); if (isLifecycleWorkPending(next)) await effects.notifyWork(tx); const [updated] = await tx.update(agents).set({ lifecycleHolds: holds, lifecycleState: next, lifecycleVersion: agent.lifecycleVersion + 1, lifecycleError: null, diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index fc0eb85522d..246c5e170d5 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,3 +1,5 @@ +import { isAgentStatusInvokable } from "@paperclipai/shared"; +import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { logger } from "../middleware/logger.js"; import type { Db } from "@paperclipai/db"; import * as lifecycle from "../modules/agent-lifecycle/index.js"; @@ -19,6 +21,11 @@ export type { LifecycleDriver, LifecycleAgent } from "../modules/agent-lifecycle export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): lifecycle.LifecycleEffects { return { notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), + async beforeStatusChange(tx, companyId, agentId, before, after) { + if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) { + await notifyNativeResponsePresentation(tx, { companyId, agentId }); + } + }, reportFailure(context, error) { const code = lifecycle.lifecycleFailureCode(error); logger.warn({ ...context, failureCode: code ?? "unclassified" }, "Agent lifecycle operation failed"); diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index fb219c6cbb1..53e8786f1a2 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -1,3 +1,4 @@ +import { notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { createFailedRetryNotices } from "./failed-retry-notices.js"; import { continuingTeamsFilePublication, type ContinuingTeamsFileTransfer } from "./microsoft-teams/file-transfer-source.js"; @@ -1564,6 +1565,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti const settle = async (status: string, result: Record) => { const [current] = await db.transaction(async (tx) => { await notifyChatActionWork(tx, "failed_run_retry"); + if (status === "processed") await notifyNativeResponsePresentation(tx, { + companyId: claimed.companyId, issueId: String(claimed.payload.issueId), + }); return tx .update(chatActions) .set({ status, result, updatedAt: new Date() }) diff --git a/server/src/services/chat-monitor-continuation.test.ts b/server/src/services/chat-monitor-continuation.test.ts index 01b094800e4..4d9c91f063e 100644 --- a/server/src/services/chat-monitor-continuation.test.ts +++ b/server/src/services/chat-monitor-continuation.test.ts @@ -1,5 +1,7 @@ +import { nativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js"; +import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { agents, agentWakeupRequests, authUsers, chatConversations, chatEndpoints, chatMessageLinks, @@ -16,14 +18,18 @@ import { createHeartbeatQueueTestDependencies } from "../__tests__/helpers/heart import { createHeartbeatQueue } from "./heartbeat/queue.js"; import { instanceSettingsService } from "./instance-settings.js"; -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MONITOR_CONTINUATION_TEST_DATABASE_URL"); +const support = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); if (!support.supported) console.warn(`Monitor publication database unavailable: ${support.reason}`); (support.supported ? describe : describe.skip)("monitor conversation continuation", () => { - let temporary: Awaited>; + let temporary: Awaited> | undefined; let db: ReturnType; beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("paperclip-monitor-publications-"); - db = createDb(temporary.connectionString); + if (external) db = createDb(external); + else { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-monitor-publications-"); + db = createDb(temporary.connectionString); + } await db.insert(authUsers).values({ id: "requester", name: "Requester", email: "requester@example.test", createdAt: new Date(), updatedAt: new Date() }); }); afterAll(async () => { await temporary?.cleanup(); }); @@ -235,7 +241,7 @@ if (!support.supported) console.warn(`Monitor publication database unavailable: expect(await f.bindings()).toEqual([]); }); - it("recovers the accepted continuation explanation after interrupted finalization without another provider result", async () => { + it.each([false, true])("recovers the accepted continuation explanation without another provider result (missing presentation: %s)", async missingPresentation => { const f = await fixture(); const [contract] = await db.insert(completionContracts).values({ companyId: f.company.id, issueId: f.issue.id, revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", @@ -272,6 +278,20 @@ if (!support.supported) console.warn(`Monitor publication database unavailable: expect(settledRun).toMatchObject({ status: "succeeded", resultJson: { finalizationPhase: "committed" } }); const priorComments = await db.select({ body: issueComments.body }).from(issueComments).where(eq(issueComments.createdByRunId, f.source.id)); expect(priorComments).toEqual([{ body: summary }]); + expect(await nativeResponsePresentationWork(db).load({ companyId: f.company.id, runId: f.source.id })).toEqual([]); + if (missingPresentation) { + await db.transaction(async tx => { + await tx.delete(chatPublications).where(eq(chatPublications.issueId, f.issue.id)); + await tx.delete(issueComments).where(eq(issueComments.createdByRunId, f.source.id)); + await tx.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'presentationDecision'` }) + .where(eq(heartbeatRuns.id, f.source.id)); + }); + const work = nativeResponsePresentationWork(db); + const entries = await work.load({ companyId: f.company.id, runId: f.source.id }); + expect(entries).toHaveLength(1); + await expect(work.check(entries[0])).resolves.toBeNull(); + expect(await work.load({ companyId: f.company.id, runId: f.source.id })).toEqual([]); + } expect(await repair()).toBe(false); expect(await repair()).toBe(false); const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.source.id)); diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 15a54e1173b..8b0d51d37fe 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { notifyChatConfirmationWork } from "./chat-confirmation-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; @@ -8,7 +9,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ChatRunMilestoneScope = { companyId: string; issueId?: string }; export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { - await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeResponsePresentation, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); + await notifyNativeResponsePresentation(tx, scope); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify({ companyId: scope.companyId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 1d8f1165fcb..a6ac545a05d 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "./native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "./native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; @@ -146,7 +146,7 @@ export async function reconcileAbandonedExecutionControl( } // A persisted result belongs to the existing finalizer, not a replacement provider. if (coordinator?.resultId) { - await notifyNativePendingEffects(tx, { companyId: run.companyId, runId: run.id }); + await notifyNativeFinalizationWork(tx, { companyId: run.companyId, runId: run.id }); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index 66a7839e25f..a00a195c39b 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -1,3 +1,4 @@ +import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { isSafeNativeChatProgressEventType } from "./safe-native-chat-progress.js"; import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; import { boundRunIssueIdSql } from "./run-task-binding.js"; @@ -72,7 +73,8 @@ export async function appendHeartbeatRunEvent( input: AppendHeartbeatRunEventInput, ): Promise { const projectsProgress = isSafeNativeChatProgressEventType(input.eventType); - if (!input.nativeSource && !input.retryExhaustion && !projectsProgress) { + const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent); + if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse) { const allocated = db.$with("allocated_run_event", {}).as(sql` update ${heartbeatRuns} set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1, @@ -101,7 +103,8 @@ export async function appendHeartbeatRunEvent( .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() } : {}) }) + ...(projectsProgress || projectsResponse ? { runtimeMode: heartbeatRuns.runtimeMode } : {}), + ...(projectsProgress ? { issueId: boundRunIssueIdSql() } : {}) }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .for("update") @@ -175,6 +178,9 @@ export async function appendHeartbeatRunEvent( if (projectsProgress && run.runtimeMode === "native" && run.issueId) { await notifyChatNativeProgress(tx, { companyId: run.companyId, issueId: run.issueId }); } + if (projectsResponse && run.runtimeMode === "native") { + await notifyNativeResponsePresentation(tx, { companyId: run.companyId, runId: input.runId }); + } const seq = await allocateHeartbeatRunEventSeq( tx as unknown as Db, input.runId, diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts new file mode 100644 index 00000000000..d1441e7ba5d --- /dev/null +++ b/server/src/services/native-finalization-notifications.ts @@ -0,0 +1,10 @@ +import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeFinalizationWorkScope = { companyId: string; runId?: string }; + +export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { + await notifyNativeResponsePresentation(transaction, scope); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativePendingEffects, + JSON.stringify({ companyId: scope.companyId, runId: scope.runId })); +} diff --git a/server/src/services/native-pending-effect-notifications.ts b/server/src/services/native-pending-effect-notifications.ts deleted file mode 100644 index e9c28d79782..00000000000 --- a/server/src/services/native-pending-effect-notifications.ts +++ /dev/null @@ -1,8 +0,0 @@ -import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; - -export type NativePendingEffectScope = { companyId: string; runId?: string }; - -export async function notifyNativePendingEffects(transaction: object, scope: NativePendingEffectScope) { - await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativePendingEffects, - JSON.stringify({ companyId: scope.companyId, runId: scope.runId })); -} diff --git a/server/src/services/native-response-presentation-notifications.ts b/server/src/services/native-response-presentation-notifications.ts new file mode 100644 index 00000000000..8be14daae95 --- /dev/null +++ b/server/src/services/native-response-presentation-notifications.ts @@ -0,0 +1,14 @@ +import { readCompletedAssistantMessageCandidate } from "./heartbeat-run-summary.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeResponsePresentationScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; + +export async function notifyNativeResponsePresentation(tx: object, scope: NativeResponsePresentationScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeResponsePresentation, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); +} + +export function isNativeResponseEvidence(eventType: string, prpEvent: unknown): boolean { + return eventType === "run.result.accepted" || eventType === "item.completed" && + readCompletedAssistantMessageCandidate({ seq: 0, prpEvent })?.channel === "final"; +} diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index 680ab867033..4243bcfa6a3 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import type { Db } from "@paperclipai/db"; import type { NativeInteractionResponseEnvelope } from "../../vendor/paperclip-runner/index.js"; @@ -633,7 +633,7 @@ export async function routePersistedNativeResultAttention(input: { const delegation = effects.find((effect) => effect.kind === "create_delegated_issue"); const finalizationError = effects.find((effect) => effect.kind === "record_finalization_error"); await input.db.transaction(async tx => { - await notifyNativePendingEffects(tx, { companyId: binding.companyId, runId: input.runId }); + await notifyNativeFinalizationWork(tx, { companyId: binding.companyId, runId: input.runId }); await tx.update(nativeRunFinalizations).set({ phase: finalizationError ? "retryable_failure" : "committed", assessmentId: existingAssessment.id, diff --git a/server/src/services/native-runtime/native-pending-effect-work.ts b/server/src/services/native-runtime/native-pending-effect-work.ts index 5893f2eadf3..3f276a0b636 100644 --- a/server/src/services/native-runtime/native-pending-effect-work.ts +++ b/server/src/services/native-runtime/native-pending-effect-work.ts @@ -2,7 +2,7 @@ import { and, asc, eq, exists, gt, inArray, isNotNull, ne } from "drizzle-orm"; import { heartbeatRuns, issues, nativeRunFinalizations, statusDecisionEffects, type Db } from "@paperclipai/db"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; -import type { NativePendingEffectScope } from "../native-pending-effect-notifications.js"; +import type { NativeFinalizationWorkScope } from "../native-finalization-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { dispatchPendingNativeStatusEffects, nativePendingEffectDueAt } from "./status-decision-committer.js"; @@ -15,7 +15,7 @@ export function pendingNativeStatusEffectsCondition(db: Db) { ))); } -function findPendingEffects(db: Db, scope?: NativePendingEffectScope, after?: string) { +function findPendingEffects(db: Db, scope?: NativeFinalizationWorkScope, after?: string) { return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, issueId: nativeRunFinalizations.issueId, decisionId: nativeRunFinalizations.decisionId, @@ -36,7 +36,7 @@ function findPendingEffects(db: Db, scope?: NativePendingEffectScope, after?: st export function registerNativePendingEffectWork(db: Db, coordinator: ReturnType, canRun: () => boolean) { type Entry = { id: string; companyId: string; dueAt: number }; - async function load(scope?: NativePendingEffectScope, signal?: AbortSignal): Promise { + async function load(scope?: NativeFinalizationWorkScope, signal?: AbortSignal): Promise { const entries: Entry[] = []; let after: string | undefined; while (canRun() && !signal?.aborted) { @@ -49,8 +49,8 @@ export function registerNativePendingEffectWork(db: Db, coordinator: ReturnType< } return registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.nativePendingEffects, batchSize: 25, - decodeScope: (hint: string): NativePendingEffectScope | undefined => hint === "*" ? undefined : JSON.parse(hint), - matches: (entry: Entry, scope?: NativePendingEffectScope) => !scope || entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId), + decodeScope: (hint: string): NativeFinalizationWorkScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativeFinalizationWorkScope) => !scope || entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId), load, check: async (entry: Entry, signal?: AbortSignal) => { if (!canRun() || signal?.aborted) throw new Error("Native pending effect recovery stopped"); diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts index bd260d1366e..243fd788ddf 100644 --- a/server/src/services/native-runtime/native-response-presentation-work.ts +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -2,17 +2,17 @@ import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, type Db } from "@paperclipai/db"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; -import type { ChatRunMilestoneScope } from "../chat-run-milestone-notifications.js"; +import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js"; import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wakeup.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; -type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; +type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; export function nativeResponsePresentationWork(db: Db) { - function candidates(scope?: ChatRunMilestoneScope, after?: string, runId?: string) { + function candidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) { return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, - issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, }).from(nativeRunFinalizations) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) @@ -46,15 +46,17 @@ export function nativeResponsePresentationWork(db: Db) { inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), ))), scope ? and(eq(heartbeatRuns.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined, + scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, + scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined, after ? gt(heartbeatRuns.id, after) : undefined, runId ? eq(heartbeatRuns.id, runId) : undefined, )).orderBy(asc(heartbeatRuns.id)).limit(100); } - async function load(scope?: ChatRunMilestoneScope, signal?: AbortSignal, canRun = () => true): Promise { + async function load(scope?: NativeResponsePresentationScope, signal?: AbortSignal, canRun = () => true): Promise { const entries: Entry[] = []; let after: string | undefined; while (canRun() && !signal?.aborted) { const rows = await candidates(scope, after); - for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); + for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); if (rows.length < 100) return entries; after = rows[rows.length - 1].id; } @@ -76,8 +78,8 @@ export function nativeResponsePresentationWork(db: Db) { function register(coordinator: ReturnType, canRun: () => boolean) { const worker = registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.nativeResponsePresentation, - decodeScope: (hint: string): ChatRunMilestoneScope | undefined => hint === "*" ? undefined : JSON.parse(hint), - matches: (entry: Entry, scope?: ChatRunMilestoneScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId), + decodeScope: (hint: string): NativeResponsePresentationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativeResponsePresentationScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId) && (!scope.agentId || entry.agentId === scope.agentId), load: (scope, signal) => load(scope, signal, canRun), check: (entry, signal) => { if (!canRun()) throw new Error("Native response presentation recovery paused"); 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 593f5c0b4b2..1f911a854b3 100644 --- a/server/src/services/native-runtime/native-run-coordinator-store.ts +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -1,4 +1,5 @@ -import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; +import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { isSafeNativeChatProgressEventType } from "../safe-native-chat-progress.js"; import { notifyChatNativeProgress } from "../chat-native-progress-notifications.js"; @@ -297,6 +298,9 @@ export class NativeRunCoordinatorStore { if (isSafeNativeChatProgressEventType(event.eventType)) { await notifyChatNativeProgress(tx, { companyId: this.#binding.companyId, issueId: this.#binding.issueId }); } + if (isNativeResponseEvidence(event.eventType, canonicalPayload)) { + await notifyNativeResponsePresentation(tx, this.#binding); + } const cursor = run.nextEventSeq; const [inserted] = await tx .insert(heartbeatRunEvents) @@ -523,7 +527,7 @@ export class NativeRunCoordinatorStore { ) { throw new Error("native_finalization_lease_conflict"); } - if (row.decisionId) await notifyNativePendingEffects(tx, this.#binding); + if (row.decisionId) await notifyNativeFinalizationWork(tx, this.#binding); if ( row.leaseOwner === input.leaseOwner && row.leaseExpiresAt && @@ -566,7 +570,7 @@ export class NativeRunCoordinatorStore { throw new Error("native_finalization_retry_invalid"); } const [updated] = await this.#db.transaction(async tx => { - await notifyNativePendingEffects(tx, this.#binding); + await notifyNativeFinalizationWork(tx, this.#binding); if (input.failureCode === "native_workspace_sync_out_unsafe_archive") { await notifyUnsafeWorkspaceExportRecovery(tx, this.#binding); } diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index dbaad2dacee..6429e083646 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -214,7 +214,7 @@ async function claimCoordinator(input: { coordinator.leaseOwner !== leaseOwner ) throw new Error("native_finalization_lease_busy"); - if (coordinator.decisionId) await notifyNativePendingEffects(tx, coordinator); + if (coordinator.decisionId) await notifyNativeFinalizationWork(tx, coordinator); const [updated] = await tx .update(nativeRunFinalizations) .set({ @@ -382,7 +382,7 @@ async function recordRetryableFailure(input: { ? "native_workspace_sync_out_retry_exhausted" : "native_finalization_retry_exhausted" : input.failureCode; - await notifyNativePendingEffects(tx, { companyId: input.run.companyId, runId: input.run.id }); + await notifyNativeFinalizationWork(tx, { companyId: input.run.companyId, runId: input.run.id }); if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 9714403b1d1..6fa55ed040e 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; @@ -3814,7 +3814,7 @@ export async function reconcileRetainedNativeSessionCleanup( renameSync(owned.root, archive); } renameSync(copy, owned.root); - await notifyNativePendingEffects(tx, { companyId: owned.run.companyId, runId: owned.run.id }); + await notifyNativeFinalizationWork(tx, { companyId: owned.run.companyId, runId: owned.run.id }); await tx .update(nativeRunFinalizations) .set({ @@ -3872,7 +3872,7 @@ export async function reconcileRetainedNativeSessionCleanup( .limit(1) .then((rows) => rows[0]); if (!current || current.leaseOwner !== leaseOwner) return; - await notifyNativePendingEffects(tx, { companyId: owned.run.companyId, runId: owned.run.id }); + await notifyNativeFinalizationWork(tx, { companyId: owned.run.companyId, runId: owned.run.id }); await tx .update(nativeRunFinalizations) .set({ @@ -9477,7 +9477,7 @@ async function executePaperclipNativeSessionWithinScope( ) .for("update"); await notifyExecutionRunWork(tx); - await notifyNativePendingEffects(tx, input.execution.binding); + await notifyNativeFinalizationWork(tx, input.execution.binding); const updated = await tx .update(nativeRunFinalizations) .set({ @@ -9813,7 +9813,7 @@ async function executePaperclipNativeSessionWithinScope( } const releaseNow = new Date(); const released = await timeController("coordinator_release", () => input.db.transaction(async tx => { - await notifyNativePendingEffects(tx, input.execution.binding); + await notifyNativeFinalizationWork(tx, input.execution.binding); return tx.update(nativeRunFinalizations) .set({ leaseOwner: null, diff --git a/server/src/services/native-runtime/native-workspace-export-recovery.ts b/server/src/services/native-runtime/native-workspace-export-recovery.ts index 6961ea01515..a86883cd4d3 100644 --- a/server/src/services/native-runtime/native-workspace-export-recovery.ts +++ b/server/src/services/native-runtime/native-workspace-export-recovery.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { alias } from "drizzle-orm/pg-core"; import type { UnsafeWorkspaceExportScope } from "../unsafe-workspace-export-notifications.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; @@ -96,7 +96,7 @@ async function recoverCandidate(db: Db, candidate: { runId: string; companyId: s phase: "workspace_finalize", status: "succeeded", exitCode: 0, finishedAt: now, metadata: { owningService: "native_workspace_finalizer", workspaceSync: { omitted: true, reason: "restore_unsafe_archive", legacy: true } }, }); - await notifyNativePendingEffects(tx, { companyId: run.companyId, runId: run.id }); + await notifyNativeFinalizationWork(tx, { companyId: run.companyId, runId: run.id }); await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, failureDetail: { workspaceFinalizeAttempt: 0, legacyUnsafeExportOmitted: { resultId: result.id, recoveredAt: now.toISOString() } }, updatedAt: now, }).where(eq(nativeRunFinalizations.runId, run.id)); diff --git a/server/src/services/native-runtime/native-workspace-export-retry.ts b/server/src/services/native-runtime/native-workspace-export-retry.ts index 69485cb4d74..99e286a17ff 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { randomUUID } from "node:crypto"; import type { Environment, EnvironmentLease } from "@paperclipai/shared"; @@ -135,7 +135,7 @@ export async function retryNativeWorkspaceExport(input: { const now = new Date(); const retry = { ...receipt, requestId: randomUUID(), actorId: input.actorId, repairNote: input.repairNote, requestedAt: now.toISOString(), stoppedProvider: initial.lease.metadata?.remoteExecutionTermination }; - await notifyNativePendingEffects(tx, { companyId: current.run.companyId, runId: current.run.id }); + await notifyNativeFinalizationWork(tx, { companyId: current.run.companyId, runId: current.run.id }); await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, failureDetail: { ...current.coordinator.failureDetail, workspaceFinalizeAttempt: 0, workspaceExportRetry: retry }, updatedAt: now }).where(eq(nativeRunFinalizations.runId, current.run.id)); await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null, diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 5321a764f8b..46700cce5fe 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -1,3 +1,5 @@ +import { nativeResponsePresentationWork } from "./native-response-presentation-work.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { asc, eq, sql } from "drizzle-orm"; import { generateKeyPairSync, randomUUID } from "node:crypto"; @@ -61,6 +63,8 @@ import { subscribeCompanyLiveEvents, } from "../live-events.js"; +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_CONTROL_PLANE_TEST_DATABASE_URL"); + describe("PaperclipControlPlanePort conformance", () => { let temporary: Awaited> | null = null; let db: ReturnType; @@ -83,8 +87,11 @@ describe("PaperclipControlPlanePort conformance", () => { const governanceRunnerId = "00000000-0000-4000-8000-000000000021"; beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-port-"); - db = createDb(temporary.connectionString); + if (external) db = createDb(external); + else { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-port-"); + db = createDb(temporary.connectionString); + } const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity; await db.insert(companies).values({ id: identity.companyId, name: "Phase 6", issuePrefix: "P6C" }); await db.insert(agents).values({ @@ -609,7 +616,7 @@ describe("PaperclipControlPlanePort conformance", () => { completionContractSha256: localContractSha, contextSnapshot: { issueId }, }); - const observerDb = createDb(temporary!.connectionString); + const observerDb = createDb(external ?? temporary!.connectionString); const observed: Array> = []; const committedCallbacks: string[] = []; let visibilityCheck: Promise> | null = null; @@ -1054,7 +1061,10 @@ describe("PaperclipControlPlanePort conformance", () => { .where(eq(heartbeatRuns.id, taskRunId)); expect(await repair()).toBe(false); await db.update(heartbeatRuns).set({ contextSnapshot: {} }).where(eq(heartbeatRuns.id, taskRunId)); - expect(await repair()).toBe(true); + const work = nativeResponsePresentationWork(db); + const [entry] = await work.load({ companyId: identity.companyId, runId: taskRunId }); + expect(entry?.id).toBe(taskRunId); + expect(await work.check(entry)).toBeNull(); expect(await repair()).toBe(false); expect(await db.select({ body: issueComments.body }).from(issueComments) .where(eq(issueComments.createdByRunId, taskRunId)).orderBy(asc(issueComments.createdAt))) diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index bf233cd933e..10a59a0a2aa 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js"; import { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js"; @@ -151,7 +151,7 @@ export async function dispatchPendingNativeStatusEffects(input: { targetId: effect.targetId, }); } - await notifyNativePendingEffects(tx, input); + await notifyNativeFinalizationWork(tx, input); await notifyChatRunMilestones(tx, input); await tx .update(statusDecisionEffects) @@ -1238,7 +1238,7 @@ async function materializeDecisionEffect(input: { ) .for("update"); if (pending.length === 0) throw new Error("native_pending_effect_missing"); - await notifyNativePendingEffects(input.tx, { companyId: input.companyId }); + await notifyNativeFinalizationWork(input.tx, { companyId: input.companyId }); await input.tx .update(statusDecisionEffects) .set({ @@ -1888,7 +1888,7 @@ export async function commitNativeStatusDecision(input: { effect => effect.kind === "record_finalization_error", ); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.companyId); - await notifyNativePendingEffects(tx, input); + await notifyNativeFinalizationWork(tx, input); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/preserved-result-finalization.ts b/server/src/services/preserved-result-finalization.ts index 083e84fefe6..69c47de6ac5 100644 --- a/server/src/services/preserved-result-finalization.ts +++ b/server/src/services/preserved-result-finalization.ts @@ -1,4 +1,4 @@ -import { notifyNativePendingEffects } from "./native-pending-effect-notifications.js"; +import { notifyNativeFinalizationWork } from "./native-finalization-notifications.js"; import { and, desc, eq, gte, ne, or, sql } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { readSystemRecovery } from "./system-recovery-state.js"; @@ -46,7 +46,7 @@ export async function resumePreservedResultFinalization(db: Db, companyId: strin const generationId = intent?.id ?? recovery.generationId; if (coordinator.recoveryHistory.some(entry => entry.kind === "preserved_result_finalization" && entry.generationId === generationId)) return false; - await notifyNativePendingEffects(tx, { companyId, runId }); + await notifyNativeFinalizationWork(tx, { companyId, runId }); await tx.update(nativeRunFinalizations).set({ phase: coordinator.assessmentId ? "arbitrating" : "workspace_finalizing", nextAttemptAt: new Date(), failureCode: null, failureDetail: null, updatedAt: new Date(),