diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 8b614c04b44..8b562d1a42e 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2858,6 +2858,12 @@ package avatar PNGs using the API renderer; static hosting needs no API proxy. ### Investigating polling load +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 +configuration, or provider results. Keep this read fresh so terminal or deleted +runs cannot publish late progress from an old executor snapshot. + Run the deterministic native event replay with the normal server test runner: ```sh diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 53092b6f8e9..c405ab8a1e5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -810,6 +810,7 @@ export function heartbeatService( getAgent, resolveSessionBeforeForWakeup, getRun, + getRunStatus, ensureRuntimeState, getTaskSession, getLatestAgentConfigRevision, @@ -1306,7 +1307,7 @@ export function heartbeatService( issuesSvc, recovery, companySkills, - getRun, + getRunStatus, emitTerminalAgentTaskRun, getIssueExecutionContext, budgets, diff --git a/server/src/services/heartbeat/run-lifecycle.test.ts b/server/src/services/heartbeat/run-lifecycle.test.ts index 7c75060ca71..a8863e7c8f7 100644 --- a/server/src/services/heartbeat/run-lifecycle.test.ts +++ b/server/src/services/heartbeat/run-lifecycle.test.ts @@ -1,3 +1,4 @@ +import { randomBytes } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; @@ -7,6 +8,7 @@ import type { CurrentUserRedactionOptions } from "../../log-redaction.js"; import { publishLiveEvent } from "../live-events.js"; import { getHeartbeatRunRuntimeStatus } from "../heartbeat-run-runtime-status.js"; import { createHeartbeatRunState } from "./run-state.js"; +import { heartbeatService } from "../heartbeat.js"; import { createHeartbeatRunPreparation } from "./run-preparation.js"; import { createHeartbeatLifecycle, persistHeartbeatRunProcessMetadata, type HeartbeatLifecycleDependencies } from "./run-lifecycle.js"; import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js"; @@ -19,7 +21,7 @@ function callbacks(db: Db) { return { ...state, ...createHeartbeatRunPreparation(db), - getRun: vi.fn(state.getRun), + getRunStatus: vi.fn(state.getRunStatus), issuesSvc: { addComment: vi.fn(), listReviewAttention: vi.fn(), @@ -84,12 +86,25 @@ describe("heartbeat lifecycle module boundary", () => { const database = guardedDatabase(); const deps = callbacks(database.db); const run = { id: "late-progress", companyId: "company", agentId: "agent", status: "running", contextSnapshot: {} } as Run; - deps.getRun.mockResolvedValue({ ...run, status: "succeeded" }); + deps.getRunStatus.mockResolvedValue({ ...run, status: "succeeded" }); expect(await createHeartbeatLifecycle(database.db, deps).recordCurrentHeartbeatRunRuntimeProgress(run, { phase: "finalize", message: "late update" }, null)).toBeNull(); - expect(deps.getRun).toHaveBeenCalledWith(run.id); + expect(deps.getRunStatus).toHaveBeenCalledWith(run.id); expect(getHeartbeatRunRuntimeStatus(run.id)).toBeNull(); expect(database.access).not.toHaveBeenCalled(); }); + + it("does not publish progress when the current status read fails", async () => { + const database = guardedDatabase(); + const deps = callbacks(database.db); + deps.getRunStatus.mockRejectedValue(new Error("status read failed")); + const run = { id: "unavailable-progress", companyId: "company", agentId: "agent", status: "running" }; + vi.mocked(publishLiveEvent).mockClear(); + await expect(createHeartbeatLifecycle(database.db, deps) + .recordCurrentHeartbeatRunRuntimeProgress(run, { phase: "restore", message: "restoring" }, null)) + .rejects.toThrow("status read failed"); + expect(publishLiveEvent).not.toHaveBeenCalled(); + expect(database.access).not.toHaveBeenCalled(); + }); }); const support = await getEmbeddedPostgresTestSupport(); @@ -118,6 +133,53 @@ describe.skipIf(!support.supported)("heartbeat lifecycle database wiring", () => return { company, agent, wake, run }; } + it.each(["succeeded", "failed", "cancelled", "timed_out", "interrupted", "deleted"])( + "rechecks progress after a run becomes %s without reading its large payloads", + async (status) => { + const { run } = await fixture({ + contextSnapshot: { prompt: randomBytes(128 * 1024).toString("hex") }, + runnerProfileJson: { configuration: randomBytes(128 * 1024).toString("hex") }, + resultJson: { providerData: randomBytes(128 * 1024).toString("hex") }, + }); + const deps = callbacks(db); + const lifecycle = createHeartbeatLifecycle(db, deps); + const update = { phase: "restore" as const, message: "Restoring workspace", currentToolName: "sync" }; + const progress = await lifecycle.recordCurrentHeartbeatRunRuntimeProgress(run, update, null); + expect(progress).toMatchObject({ + companyId: run.companyId, agentId: run.agentId, runId: run.id, + phase: update.phase, message: update.message, currentToolName: "sync", + }); + expect(await deps.getRunStatus.mock.results[0].value).toEqual({ + id: run.id, companyId: run.companyId, agentId: run.agentId, status: "running", + }); + expect(publishLiveEvent).toHaveBeenCalledTimes(1); + expect(publishLiveEvent).toHaveBeenCalledWith(expect.objectContaining({ + companyId: run.companyId, + type: "heartbeat.run.progress", + payload: expect.objectContaining({ runId: run.id, agentId: run.agentId, phase: update.phase, message: update.message }), + })); + + if (status === "deleted") await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + else await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, run.id)); + + expect(await lifecycle.recordCurrentHeartbeatRunRuntimeProgress(run, update, null)).toBeNull(); + expect(deps.getRunStatus).toHaveBeenCalledTimes(2); + expect(getHeartbeatRunRuntimeStatus(run.id)).toBeNull(); + expect(publishLiveEvent).toHaveBeenCalledTimes(1); + }, + ); + + it("wires fresh progress status reads through the public heartbeat service", async () => { + const { run } = await fixture(); + const service = heartbeatService(db); + const update = { phase: "config_sync" as const, message: "Syncing workspace" }; + expect(await service.recordRuntimeProgress(run, update, null)).toMatchObject({ runId: run.id, message: update.message }); + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, run.id)); + expect(await service.recordRuntimeProgress(run, update, null)).toBeNull(); + expect(getHeartbeatRunRuntimeStatus(run.id)).toBeNull(); + expect(publishLiveEvent).toHaveBeenCalledTimes(1); + }); + it("preserves sandbox process birth instead of inspecting a colliding controller PID", async () => { const { run } = await fixture(); const remoteBirth = "2026-10-09T00:00:00.000Z"; diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 0ee088a4d51..5f299e855fa 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -193,7 +193,7 @@ type HeartbeatRun = typeof heartbeatRuns.$inferSelect; /** Lifecycle owns run writes; the service supplies reporting and admission effects. */ export interface HeartbeatLifecycleDependencies extends Pick, - "getRun" | "resolveSessionBeforeForWakeup" | "ensureRuntimeState">, + "getRunStatus" | "resolveSessionBeforeForWakeup" | "ensureRuntimeState">, Pick, "getIssueExecutionContext" | "resolveResponsibleUserIdForRunContext">, Pick { @@ -388,7 +388,7 @@ export function publishHeartbeatRunRuntimeProgress(status: { function recordHeartbeatRunRuntimeProgress( run: Pick< typeof heartbeatRuns.$inferSelect, - "id" | "companyId" | "agentId" | "status" | "contextSnapshot" + "id" | "companyId" | "agentId" | "status" >, update: RuntimeStatusUpdate, issueId: string | null, @@ -635,7 +635,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl issuesSvc, recovery, companySkills, - getRun, + getRunStatus, emitTerminalAgentTaskRun, getIssueExecutionContext, budgets, @@ -992,7 +992,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl async function recordCurrentHeartbeatRunRuntimeProgress( run: Pick< typeof heartbeatRuns.$inferSelect, - "id" | "companyId" | "agentId" | "status" | "contextSnapshot" + "id" | "companyId" | "agentId" | "status" >, update: RuntimeStatusUpdate, issueId: string | null, @@ -1002,7 +1002,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl return null; } - const currentRun = await getRun(run.id); + const currentRun = await getRunStatus(run.id); if (!currentRun || !isHeartbeatRunRuntimeStatusActive(currentRun.status)) { clearHeartbeatRunRuntimeStatus(run.id); return null; diff --git a/server/src/services/heartbeat/run-state.test.ts b/server/src/services/heartbeat/run-state.test.ts index 087931f01e8..4a729328027 100644 --- a/server/src/services/heartbeat/run-state.test.ts +++ b/server/src/services/heartbeat/run-state.test.ts @@ -92,6 +92,22 @@ describe("heartbeat run state module", () => { expect(raw.columns[0].resultJson).toBe(heartbeatRuns.resultJson); }); + it("loads current run status without large fields or an encoding probe", async () => { + const binding = { id: "run", companyId: "company", agentId: "agent", status: "running" }; + const reader = readingDatabase("SQL_ASCII", [binding], []); + const state = extracted.createHeartbeatRunState(reader.db); + expect(await state.getRunStatus("run")).toEqual(binding); + expect(await state.getRunStatus("missing")).toBeNull(); + expect(reader.execute).not.toHaveBeenCalled(); + expect(reader.columns).toEqual(Array(2).fill({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + })); + expect(reader.scopes).toEqual([["run"], ["missing"]]); + }); + it("loads run read bindings without large fields or an encoding probe", async () => { const binding = { id: "run", companyId: "company", scopeKind: "issue", issueId: "issue" }; const reader = readingDatabase("SQL_ASCII", [binding], []); diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index a6f302ab64a..30f9ef0b001 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -1119,6 +1119,19 @@ export function createHeartbeatRunState(db: Db) { .then((rows) => rows[0] ?? null); } + async function getRunStatus(runId: string) { + return db + .select({ + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + } + async function getRunReadAccess(runId: string) { return db .select(heartbeatRunReadAccessColumns) @@ -1684,6 +1697,7 @@ export function createHeartbeatRunState(db: Db) { clearTaskSessions, resolveExplicitResumeSessionOverride, hasResolvablePriorSessionWorkspaceForWake, + getRunStatus, getRunReadAccess, getRunLogAccess,