diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index fd25491f0d4..a8c500a4387 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2871,6 +2871,13 @@ package avatar PNGs using the API renderer; static hosting needs no API proxy. ### Investigating polling load +Deferred-message recovery reads run admission fields before calling the release +policy. The release coordinator reads authentication evidence and then refreshes +completion identity after post-commit effects. These checks do not load stored +prompts, Runner profiles, or output excerpts. Full execution and detail reads +retain those fields. Queue cadence, delivery admission, and cleanup ownership +remain unchanged. + Runtime progress updates recheck the persisted run status before publication. This check selects only the run ID, company ID, agent ID, and status through `createHeartbeatRunState().getRunStatus`. It does not fetch prompts, runner diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 9af4a5ac956..5a31b5ee131 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -161,6 +161,9 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { it.each(["legacy", "native"])("resumes an accepted non-assignee queue through its %s owner", async runtimeMode => { const { run, pending } = await strandedMention(runtimeMode); + await db.update(heartbeatRuns).set({ contextSnapshot: { ...run.contextSnapshot, prompt: "large context".repeat(20_000) }, + runnerProfileJson: { instructions: "large profile".repeat(20_000) }, stdoutExcerpt: "large output".repeat(20_000), + }).where(eq(heartbeatRuns.id, run.id)); const deps = callbacks(db); await createHeartbeatQueue(db, deps).resumeQueuedRuns(); expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledExactlyOnceWith( @@ -170,6 +173,11 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { status: "deferred_issue_execution", runId: null, payload: pending.payload, }); expect(deps.executeRun).not.toHaveBeenCalled(); + const selected = deps.releaseIssueExecutionAndPromote.mock.calls[0][0]; + expect(Buffer.byteLength(JSON.stringify(selected))).toBeLessThan(512); + expect(selected).not.toHaveProperty("contextSnapshot"); + expect(selected).not.toHaveProperty("runnerProfileJson"); + expect(selected).not.toHaveProperty("stdoutExcerpt"); }); it("preserves a former owner's ordinary messages for the assignment handoff", async () => { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index c6ac63b6a53..515bd683fb3 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1837,7 +1837,15 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende for (const { wake } of strandedQueues) { if (!queuedCommentIdsFromWakePayload(wake.payload).length && !await readQueuedInteractionResponse(db, wake.companyId, String(wake.payload?.issueId), wake.payload)) continue; - const [latest] = await db.select().from(heartbeatRuns).where(and( + const [latest] = await db.select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + status: heartbeatRuns.status, + finishedAt: heartbeatRuns.finishedAt, + startedAt: heartbeatRuns.startedAt, + runtimeMode: heartbeatRuns.runtimeMode, + errorCode: heartbeatRuns.errorCode, + }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, wake.companyId), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${String(wake.payload?.issueId)}`, or(isNull(heartbeatRuns.nativeIssueId), sql`${heartbeatRuns.nativeIssueId}::text = ${String(wake.payload?.issueId)}`), diff --git a/server/src/services/heartbeat/run-control-reads.test.ts b/server/src/services/heartbeat/run-control-reads.test.ts new file mode 100644 index 00000000000..eb0248cc46d --- /dev/null +++ b/server/src/services/heartbeat/run-control-reads.test.ts @@ -0,0 +1,87 @@ +import { randomBytes } from "node:crypto"; +import type { Db } from "@paperclipai/db"; +import { heartbeatRuns, issues } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { describe, expect, it, vi } from "vitest"; +import { isAiAuthenticationBlocked } from "../ai-auth-failure.js"; +import { settleSlackConversation } from "../slack-conversation-lifecycle.js"; +import { createHeartbeatRunControl } from "./run-control.js"; +import { callbacks, runControlPostgresSupport, setupRunControlDatabase } from "./__tests__/run-control-fixtures.js"; + +vi.mock("../local-service-supervisor.js", async importOriginal => ({ + ...await importOriginal(), + terminateLocalService: vi.fn(async () => undefined), +})); +vi.mock("../slack-conversation-lifecycle.js", () => ({ settleSlackConversation: vi.fn(async () => undefined) })); + +describe.skipIf(!runControlPostgresSupport.supported)("execution release reads", () => { + let db: Db; + const { fixture } = setupRunControlDatabase("heartbeat-release-reads-", value => { db = value; }); + + function recordRunReads() { + const reads: unknown[][] = []; + const database = new Proxy(db, { get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields); + const from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const query = from(...args); + if (args[0] === heartbeatRuns) { + const execute = query.execute.bind(query); + query.execute = async (...executeArgs: Parameters) => { + const rows = await execute(...executeArgs); + reads.push(rows); + return rows; + }; + } + return query; + }) as typeof builder.from; + return builder; + }; + }}); + return { database, reads }; + } + + it.each([ + { errorCode: "codex_auth_required", livenessState: "blocked", resultJson: null }, + { errorCode: "codex_auth_required", livenessState: "advanced", resultJson: null }, + { errorCode: "configuration_incomplete", livenessState: "blocked", resultJson: { configurationIncomplete: { reason: "ai_connection_unavailable" } } }, + { errorCode: "configuration_incomplete", livenessState: "blocked", resultJson: { configurationIncomplete: { reason: "secret_binding_missing", agentId: "owner", missingBindings: [{ bindingType: "user_secret_ref", consumerType: "agent", consumerId: "owner", errorCode: "user_secret_missing", envKey: "OPENAI_API_KEY", configPath: "env.OPENAI_API_KEY" }] } } }, + { errorCode: "configuration_incomplete", livenessState: "blocked", resultJson: { configurationIncomplete: { reason: "workspace_unavailable" } } }, + ])("preserves authentication admission for %j without loading execution payloads", async failure => { + const { company, agent } = await fixture(); + const payload = randomBytes(128 * 1024).toString("base64"); + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + status: "failed", ...failure, contextSnapshot: { prompt: payload }, runnerProfileJson: { prompt: payload }, stdoutExcerpt: payload, + }).returning(); + const { database, reads } = recordRunReads(); + const deps = callbacks(database); + await createHeartbeatRunControl(database, deps).releaseIssueExecutionAndPromote(run); + expect(deps.wakeQueue.releaseIssueExecution).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + companyId: company.id, runId: run.id, suppressImmediateRecovery: isAiAuthenticationBlocked(run), + })); + expect(deps.getRun).not.toHaveBeenCalled(); + expect(reads).toHaveLength(2); + expect(Buffer.byteLength(JSON.stringify(reads))).toBeLessThan(2048); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(stored).toEqual(run); + }); + + it("reads the completed identity after post-commit effects and preserves the full record", async () => { + const { company, agent } = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Release destination" }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + status: "running", contextSnapshot: { issueId: 42, taskId: "", prompt: "retained" }, + }).returning(); + const deps = callbacks(db); + vi.spyOn(deps, "applyWakeQueuePostCommitEffects").mockImplementationOnce(async () => { + await db.update(heartbeatRuns).set({ status: "succeeded", contextSnapshot: { issueId: 42, taskId: issue.id, prompt: "retained" } }).where(eq(heartbeatRuns.id, run.id)); + }); + vi.mocked(settleSlackConversation).mockClear(); + await createHeartbeatRunControl(db, deps).releaseIssueExecutionAndPromote(run, { resumeStrandedQueue: true }); + expect(settleSlackConversation).toHaveBeenCalledExactlyOnceWith(db, company.id, issue.id); + expect(deps.wakeQueue.releaseIssueExecution).toHaveBeenCalledWith(expect.objectContaining({ resumeStrandedQueue: true })); + expect(deps.getRun).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index e9b21a98aa3..7f80f99321a 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -815,7 +815,15 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont options: { suppressImmediateRecovery?: boolean; resumeStrandedQueue?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] } = {}, ) { try { - const source = await getRun(run.id); + const [source] = await db.select({ + errorCode: heartbeatRuns.errorCode, + livenessState: heartbeatRuns.livenessState, + resultJson: sql>`jsonb_build_object( + 'configurationIncomplete', ${heartbeatRuns.resultJson}->'configurationIncomplete' + )`, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), + )); const sourceHandoff = await pendingCaseHandoffForRun(db, run.id, run.companyId); const { postCommitEffects } = await wakeQueue.releaseIssueExecution({ companyId: run.companyId, @@ -831,9 +839,18 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont } else { await applyWakeQueuePostCommitEffects(postCommitEffects); } - const completed = await getRun(run.id); - const issueId = readNonEmptyString(completed?.contextSnapshot?.issueId) - ?? readNonEmptyString(completed?.contextSnapshot?.taskId) ?? completed?.nativeIssueId; + const [completed] = await db.select({ + id: heartbeatRuns.id, + status: heartbeatRuns.status, + startedAt: heartbeatRuns.startedAt, + nativeIssueId: heartbeatRuns.nativeIssueId, + contextIssueId: sql`${heartbeatRuns.contextSnapshot}->'issueId'`, + contextTaskId: sql`${heartbeatRuns.contextSnapshot}->'taskId'`, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), + )); + const issueId = readNonEmptyString(completed?.contextIssueId) + ?? readNonEmptyString(completed?.contextTaskId) ?? completed?.nativeIssueId; if (sourceHandoff && completed?.status === "cancelled" && !completed.startedAt && issueId) { // A cancelled queue entry has no executeRun.finally. The canonical // bootstrap proof still needs to release and wake its durable handoff.