diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index 4bf1091a0b..beb0d4dbdf 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -8,7 +8,7 @@ import { } from "./sandbox-managed-runtime.js"; import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js"; import type { RunProcessResult } from "./server-utils.js"; -import type { RuntimeProgressSink } from "./runtime-progress.js"; +import type { RuntimeProgressSink, RuntimeStatusSink } from "./runtime-progress.js"; export interface CommandManagedRuntimeRunner { /** @@ -242,6 +242,7 @@ export async function prepareCommandManagedRuntime(input: { // Upload progress sink. Forwarded to prepareSandboxManagedRuntime; the child // task wires it into the byte-counting writeFile/readFile transport. onProgress?: RuntimeProgressSink; + onRuntimeProgress?: RuntimeStatusSink; }): Promise { const timeoutMs = input.spec.timeoutMs && input.spec.timeoutMs > 0 ? input.spec.timeoutMs : 300_000; const workspaceRemoteDir = input.workspaceRemoteDir ?? input.spec.remoteCwd; @@ -290,6 +291,7 @@ export async function prepareCommandManagedRuntime(input: { preserveAbsentOnRestore: input.preserveAbsentOnRestore, assets: input.assets, onProgress: input.onProgress, + onRuntimeProgress: input.onRuntimeProgress, }); } } @@ -325,5 +327,6 @@ export async function prepareCommandManagedRuntime(input: { preserveAbsentOnRestore: input.preserveAbsentOnRestore, assets: input.assets, onProgress: input.onProgress, + onRuntimeProgress: input.onRuntimeProgress, }); } diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 86d3f1a8bc..6978e8e903 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -28,7 +28,7 @@ import { } from "./server-utils.js"; import { sanitizeRemoteExecutionEnv } from "./remote-execution-env.js"; import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js"; -import type { RuntimeProgressSink } from "./runtime-progress.js"; +import type { RuntimeProgressSink, RuntimeStatusSink } from "./runtime-progress.js"; export type { RuntimeProgressSink } from "./runtime-progress.js"; @@ -83,6 +83,7 @@ export interface AdapterExecutionTargetProcessOptions { timeoutSec: number; graceSec: number; onLog: (stream: "stdout" | "stderr", chunk: string) => Promise; + onRuntimeProgress?: RuntimeStatusSink; onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise; terminalResultCleanup?: TerminalResultCleanupOptions; } @@ -407,6 +408,10 @@ export async function runAdapterExecutionTargetProcess( if (target?.kind === "remote" && target.transport === "sandbox") { const runner = requireSandboxRunner(target); const env = sanitizeRemoteExecutionEnv(options.env); + await options.onRuntimeProgress?.({ + phase: "adapter_startup", + message: "Starting adapter in sandbox", + }); return await runner.execute({ command, args, @@ -938,6 +943,7 @@ export async function prepareAdapterExecutionTargetRuntime(input: { // forwarded down to the transport so the sandbox/SSH children can attach byte // counters without further changes here. onProgress?: RuntimeProgressSink; + onRuntimeProgress?: RuntimeStatusSink; }): Promise { const target = input.target ?? { kind: "local" as const }; if (target.kind === "local") { @@ -990,6 +996,7 @@ export async function prepareAdapterExecutionTargetRuntime(input: { installCommand: input.installCommand, detectCommand: input.detectCommand, onProgress: input.onProgress, + onRuntimeProgress: input.onRuntimeProgress, }); return { target, diff --git a/packages/adapter-utils/src/index.ts b/packages/adapter-utils/src/index.ts index 303c028ed2..32e890e0d5 100644 --- a/packages/adapter-utils/src/index.ts +++ b/packages/adapter-utils/src/index.ts @@ -69,6 +69,9 @@ export type { RuntimeProgressTarget, RuntimeProgressReporter, RuntimeProgressReporterOptions, + RuntimeStatusPhase, + RuntimeStatusSink, + RuntimeStatusUpdate, } from "./runtime-progress.js"; export { inferOpenAiCompatibleBiller } from "./billing.js"; // Keep the root adapter-utils entry browser-safe because the UI imports it. diff --git a/packages/adapter-utils/src/runtime-progress.ts b/packages/adapter-utils/src/runtime-progress.ts index 8cdd7ce4ec..11beabda63 100644 --- a/packages/adapter-utils/src/runtime-progress.ts +++ b/packages/adapter-utils/src/runtime-progress.ts @@ -21,6 +21,21 @@ export type RuntimeProgressDirection = "to" | "from"; export type RuntimeProgressTarget = "sandbox" | "ssh"; +export type RuntimeStatusPhase = + | "git_sync" + | "config_sync" + | "adapter_startup" + | "restore" + | "export" + | "finalize"; + +export interface RuntimeStatusUpdate { + phase: RuntimeStatusPhase; + message: string; +} + +export type RuntimeStatusSink = (update: RuntimeStatusUpdate) => void | Promise; + export interface RuntimeProgressReporterOptions { sink: RuntimeProgressSink; phase: RuntimeProgressPhase; diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts index 803071b754..2681928d74 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts @@ -103,6 +103,7 @@ describe("sandbox managed runtime", () => { }); }, }; + const runtimeStatuses: string[] = []; const prepared = await prepareSandboxManagedRuntime({ spec: { @@ -118,6 +119,9 @@ describe("sandbox managed runtime", () => { workspaceLocalDir: localWorkspaceDir, workspaceExclude: [".claude"], preserveAbsentOnRestore: [".claude"], + onRuntimeProgress: async (status) => { + runtimeStatuses.push(`${status.phase}:${status.message}`); + }, assets: [{ key: "skills", localDir: localAssetsDir, @@ -143,6 +147,12 @@ describe("sandbox managed runtime", () => { await expect(readFile(path.join(localWorkspaceDir, "local-stale.txt"), "utf8")).resolves.toBe("remove\n"); await expect(readFile(path.join(localWorkspaceDir, ".claude", "settings.json"), "utf8")).resolves.toBe("{\"local\":true}\n"); await expect(readFile(path.join(localWorkspaceDir, ".paperclip-runtime", "state.json"), "utf8")).resolves.toBe("{}\n"); + expect(runtimeStatuses).toEqual([ + "config_sync:Syncing workspace to sandbox", + "config_sync:Syncing runtime assets to sandbox", + "restore:Restoring workspace from sandbox", + "finalize:Finalizing sandbox workspace", + ]); }); it("syncs git-backed workspaces through a shallow standalone clone and keeps .git out of archives", async () => { @@ -197,6 +207,7 @@ describe("sandbox managed runtime", () => { await execFile("sh", ["-c", command], { maxBuffer: 32 * 1024 * 1024 }); }, }; + const runtimeStatusPhases: string[] = []; const prepared = await prepareSandboxManagedRuntime({ spec: { @@ -210,6 +221,9 @@ describe("sandbox managed runtime", () => { adapterKey: "test-adapter", client, workspaceLocalDir: localWorkspaceDir, + onRuntimeProgress: async (status) => { + runtimeStatusPhases.push(status.phase); + }, }); expect((await lstat(path.join(remoteWorkspaceDir, ".git"))).isDirectory()).toBe(true); @@ -254,6 +268,13 @@ describe("sandbox managed runtime", () => { const downloadMembers = await listTarMembers(rootDir, "workspace-download-list.tar", downloadedTars[0]!.bytes); expect(downloadMembers.some((entry) => entry === ".git" || entry.startsWith(".git/"))).toBe(false); expect(downloadMembers.some((entry) => entry === "node_modules" || entry.startsWith("node_modules/"))).toBe(false); + expect(runtimeStatusPhases).toEqual([ + "git_sync", + "config_sync", + "export", + "restore", + "finalize", + ]); }); it("excludes unignored dependency trees from git-backed workspace overlay archives", async () => { diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.ts b/packages/adapter-utils/src/sandbox-managed-runtime.ts index cf1ce0739c..7c763f42b8 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.ts @@ -20,6 +20,8 @@ import { type RuntimeProgressDirection, type RuntimeProgressPhase, type RuntimeProgressSink, + type RuntimeStatusPhase, + type RuntimeStatusSink, } from "./runtime-progress.js"; import { isRelativePathOrDescendant, shouldExcludePath } from "./exclude-patterns.js"; @@ -320,6 +322,15 @@ function tarExcludeFlags(exclude: string[] | undefined): string { return ["._*", ...(exclude ?? [])].map((entry) => `--exclude ${shellQuote(entry)}`).join(" "); } +async function emitRuntimeStatus( + sink: RuntimeStatusSink | undefined, + phase: RuntimeStatusPhase, + message: string, +): Promise { + if (!sink) return; + await Promise.resolve(sink({ phase, message })).catch(() => undefined); +} + function mergeExcludes(...groups: Array): string[] { return [...new Set(groups.flatMap((group) => group ?? []))]; } @@ -384,6 +395,7 @@ export async function prepareSandboxManagedRuntime(input: { // Upload progress sink. Threaded for the byte-counting transport rewrite; the // child task wires it into writeFile/readFile. onProgress?: RuntimeProgressSink; + onRuntimeProgress?: RuntimeStatusSink; }): Promise { const workspaceRemoteDir = input.workspaceRemoteDir ?? input.spec.remoteCwd; const runtimeRootDir = path.posix.join(workspaceRemoteDir, ".paperclip-runtime", input.adapterKey); @@ -414,6 +426,7 @@ export async function prepareSandboxManagedRuntime(input: { ...(input.preserveAbsentOnRestore ?? []), ]); if (gitSnapshot) { + await emitRuntimeStatus(input.onRuntimeProgress, "git_sync", "Syncing git history to sandbox"); await withShallowGitWorkspaceClone({ localDir: input.workspaceLocalDir, snapshot: gitSnapshot, @@ -444,6 +457,7 @@ export async function prepareSandboxManagedRuntime(input: { const workspaceTarPath = path.join(tempDir, "workspace.tar"); const workspaceArchiveDir = gitSnapshot ? path.join(tempDir, "workspace-overlay") : input.workspaceLocalDir; + await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing workspace to sandbox"); if (gitSnapshot) { await copySelectedWorkspaceEntries({ sourceDir: input.workspaceLocalDir, @@ -489,6 +503,7 @@ export async function prepareSandboxManagedRuntime(input: { } for (const asset of input.assets ?? []) { + await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing runtime assets to sandbox"); const assetTarPath = path.join(tempDir, `${asset.key}.tar`); await createTarballFromDirectory({ localDir: asset.localDir, @@ -531,6 +546,7 @@ export async function prepareSandboxManagedRuntime(input: { let importedHead: string | null = null; try { if (gitSnapshot) { + await emitRuntimeStatus(input.onRuntimeProgress, "export", "Exporting git changes from sandbox"); importedRef = createImportedGitRef("sandbox"); const remoteGitBundle = path.posix.join(runtimeRootDir, "git-delta.bundle"); const exportRef = createRemoteGitExportRef("sandbox"); @@ -559,6 +575,7 @@ export async function prepareSandboxManagedRuntime(input: { } const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-download.tar"); + await emitRuntimeStatus(input.onRuntimeProgress, "restore", "Restoring workspace from sandbox"); await input.client.run( `sh -c ${shellQuote( `mkdir -p ${shellQuote(runtimeRootDir)} && ` + @@ -593,6 +610,7 @@ export async function prepareSandboxManagedRuntime(input: { : undefined, }); } finally { + await emitRuntimeStatus(input.onRuntimeProgress, "finalize", "Finalizing sandbox workspace"); if (importedRef) { await deleteLocalGitRef({ localDir: input.workspaceLocalDir, ref: importedRef }); } diff --git a/packages/adapter-utils/src/types.ts b/packages/adapter-utils/src/types.ts index 668dab92c0..80f5f92f86 100644 --- a/packages/adapter-utils/src/types.ts +++ b/packages/adapter-utils/src/types.ts @@ -4,6 +4,7 @@ import type { SshRemoteExecutionSpec } from "./ssh.js"; import type { AdapterExecutionTarget } from "./execution-target.js"; +import type { RuntimeStatusSink } from "./runtime-progress.js"; export interface AdapterAgent { id: string; @@ -136,6 +137,7 @@ export interface AdapterExecutionContext { }; onLog: (stream: "stdout" | "stderr", chunk: string) => Promise; onMeta?: (meta: AdapterInvocationMeta) => Promise; + onRuntimeProgress?: RuntimeStatusSink; onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise; authToken?: string; } diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 73666f8ff6..ad9136d96a 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -487,6 +487,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), + onRuntimeProgress: ctx.onRuntimeProgress, assets: [ { key: "skills", @@ -790,6 +791,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), + onRuntimeProgress: ctx.onRuntimeProgress, assets: [ { key: "home", @@ -875,6 +876,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { if (stream === "stdout") { monitor?.noteStdoutChunk(chunk); diff --git a/packages/adapters/cursor-local/src/server/execute.ts b/packages/adapters/cursor-local/src/server/execute.ts index 40c3e8ff53..a5600f1df8 100644 --- a/packages/adapters/cursor-local/src/server/execute.ts +++ b/packages/adapters/cursor-local/src/server/execute.ts @@ -365,6 +365,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), + onRuntimeProgress: ctx.onRuntimeProgress, assets: [{ key: "skills", localDir: localSkillsDir, @@ -636,6 +637,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { if (stream !== "stdout") { await onLog(stream, chunk); diff --git a/packages/adapters/gemini-local/src/server/execute.ts b/packages/adapters/gemini-local/src/server/execute.ts index 4f9d54868f..561263bf46 100644 --- a/packages/adapters/gemini-local/src/server/execute.ts +++ b/packages/adapters/gemini-local/src/server/execute.ts @@ -341,6 +341,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), + onRuntimeProgress: ctx.onRuntimeProgress, assets: [{ key: "skills", localDir: localSkillsDir, @@ -591,6 +592,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), + onRuntimeProgress: ctx.onRuntimeProgress, }); restoreRemoteWorkspace = () => preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line)); @@ -474,6 +475,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), + onRuntimeProgress: ctx.onRuntimeProgress, assets: [ { key: "skills", @@ -603,6 +604,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), + onRuntimeProgress: ctx.onRuntimeProgress, assets: [ { key: "skills", @@ -709,6 +710,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise ({ const mockHeartbeatService = vi.hoisted(() => ({ buildRunOutputSilence: vi.fn(), + decorateActiveRunStatus: vi.fn(), getRunIssueSummary: vi.fn(), getActiveRunIssueSummaryForAgent: vi.fn(), getRunLogAccess: vi.fn(), @@ -194,6 +195,11 @@ describe("agent live run routes", () => { }); mockInstanceSettingsService.listCompanyIds.mockResolvedValue(["company-1"]); mockHeartbeatService.buildRunOutputSilence.mockResolvedValue(null); + mockHeartbeatService.decorateActiveRunStatus.mockImplementation((run) => ({ + ...run, + currentStatusMessage: null, + currentStatusUpdatedAt: null, + })); mockHeartbeatService.getRunIssueSummary.mockResolvedValue({ id: "run-1", status: "running", @@ -256,6 +262,8 @@ describe("agent live run routes", () => { agentName: "Builder", adapterType: "codex_local", outputSilence: null, + currentStatusMessage: null, + currentStatusUpdatedAt: null, }); expect(res.body).not.toHaveProperty("resultJson"); expect(res.body).not.toHaveProperty("contextSnapshot"); @@ -303,6 +311,29 @@ describe("agent live run routes", () => { }); }); + it("includes ephemeral current status fields on active run polling", async () => { + mockHeartbeatService.decorateActiveRunStatus.mockImplementation((run) => ({ + ...run, + currentStatusMessage: "Syncing workspace to sandbox", + currentStatusUpdatedAt: new Date("2026-04-10T09:30:05.000Z"), + })); + + const res = await requestApp( + await createApp(), + (baseUrl) => request(baseUrl).get("/api/issues/PC1A2-1295/active-run"), + ); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(mockHeartbeatService.decorateActiveRunStatus).toHaveBeenCalledWith( + expect.objectContaining({ id: "run-1", issueId: "issue-1" }), + { companyId: "company-1", issueId: "issue-1" }, + ); + expect(res.body).toMatchObject({ + currentStatusMessage: "Syncing workspace to sandbox", + currentStatusUpdatedAt: "2026-04-10T09:30:05.000Z", + }); + }); + it("uses narrow run log metadata lookups for log polling", async () => { const res = await requestApp( await createApp(), diff --git a/server/src/__tests__/heartbeat-runtime-state.test.ts b/server/src/__tests__/heartbeat-runtime-state.test.ts index 5a8faa9a55..e6c182b3ce 100644 --- a/server/src/__tests__/heartbeat-runtime-state.test.ts +++ b/server/src/__tests__/heartbeat-runtime-state.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; import { agents, @@ -14,7 +14,23 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -import { heartbeatService } from "../services/heartbeat.ts"; +import { subscribeCompanyLiveEvents } from "../services/live-events.ts"; +import { + clearAllHeartbeatRunRuntimeStatuses, + getHeartbeatRunRuntimeStatus, +} from "../services/heartbeat-run-runtime-status.ts"; + +vi.doMock("../adapters/index.js", () => ({ + getServerAdapter: vi.fn(() => ({ + type: "process", + execute: vi.fn(), + testEnvironment: vi.fn(), + })), + listAdapterModelProfiles: vi.fn(() => []), + runningProcesses: new Map(), +})); + +const { heartbeatService } = await import("../services/heartbeat.ts"); const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -35,6 +51,7 @@ describeEmbeddedPostgres("heartbeat runtime state deduplication", () => { }, 20_000); afterEach(async () => { + clearAllHeartbeatRunRuntimeStatuses(); await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); @@ -85,4 +102,168 @@ describeEmbeddedPostgres("heartbeat runtime state deduplication", () => { stateJson: {}, }); }); + + it("publishes runtime progress without persisting heartbeat run events", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const runId = randomUUID(); + const issueId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "running", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + const [insertedRun] = await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + status: "running", + contextSnapshot: { issueId }, + }).returning(); + const run = insertedRun!; + + const liveEvents: unknown[] = []; + const unsubscribe = subscribeCompanyLiveEvents(companyId, (event) => { + liveEvents.push(event); + }); + try { + const heartbeat = heartbeatService(db); + const status = await heartbeat.recordRuntimeProgress(run, { + phase: "config_sync", + message: "Syncing workspace to sandbox", + }, issueId); + + expect(status).toMatchObject({ + companyId, + issueId, + agentId, + runId, + phase: "config_sync", + message: "Syncing workspace to sandbox", + }); + expect(heartbeat.decorateActiveRunStatus({ + id: runId, + companyId, + agentId, + issueId, + status: "running", + })).toMatchObject({ + currentStatusMessage: "Syncing workspace to sandbox", + }); + expect(liveEvents).toContainEqual(expect.objectContaining({ + companyId, + type: "heartbeat.run.progress", + payload: expect.objectContaining({ + runId, + agentId, + issueId, + phase: "config_sync", + message: "Syncing workspace to sandbox", + }), + })); + + const persistedEvents = await db.select().from(heartbeatRunEvents); + expect(persistedEvents).toHaveLength(0); + + await heartbeat.cancelRun(runId, "test cleanup"); + expect(getHeartbeatRunRuntimeStatus(runId)).toBeNull(); + } finally { + unsubscribe(); + } + }); + + it("ignores late runtime progress after the persisted run is terminal", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const runId = randomUUID(); + const issueId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "running", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + const [insertedRun] = await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + status: "running", + contextSnapshot: { issueId }, + }).returning(); + const staleRunningRun = insertedRun!; + + const liveEvents: unknown[] = []; + const unsubscribe = subscribeCompanyLiveEvents(companyId, (event) => { + liveEvents.push(event); + }); + try { + const heartbeat = heartbeatService(db); + await heartbeat.recordRuntimeProgress(staleRunningRun, { + phase: "config_sync", + message: "Syncing workspace to sandbox", + }, issueId); + + expect(getHeartbeatRunRuntimeStatus(runId)).toMatchObject({ + runId, + phase: "config_sync", + }); + + liveEvents.length = 0; + await db + .update(heartbeatRuns) + .set({ + status: "succeeded", + finishedAt: new Date("2026-06-24T00:01:00.000Z"), + updatedAt: new Date("2026-06-24T00:01:00.000Z"), + }) + .where(eq(heartbeatRuns.id, runId)); + + const lateStatus = await heartbeat.recordRuntimeProgress(staleRunningRun, { + phase: "finalize", + message: "Finalizing sandbox workspace", + }, issueId); + + expect(lateStatus).toBeNull(); + expect(getHeartbeatRunRuntimeStatus(runId)).toBeNull(); + expect(liveEvents).not.toContainEqual(expect.objectContaining({ + type: "heartbeat.run.progress", + })); + expect(await db.select().from(heartbeatRunEvents)).toHaveLength(0); + } finally { + unsubscribe(); + } + }); }); diff --git a/server/src/index.ts b/server/src/index.ts index 1e6ac16bcf..331488e985 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -852,6 +852,14 @@ export async function startServer(): Promise { }); setInterval(() => { + const sweptRuntimeStatuses = heartbeat.sweepExpiredRuntimeStatuses(); + if (sweptRuntimeStatuses > 0) { + logger.info( + { swept: sweptRuntimeStatuses }, + "heartbeat runtime-status sweeper cleared expired entries", + ); + } + void heartbeat .tickTimers(new Date()) .then((result) => { diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index ea16d1b37a..cd92409781 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -3517,14 +3517,14 @@ export function agentRoutes( const rows = [...liveRuns, ...recentRuns]; res.json(await Promise.all(rows.map(async (run) => ({ - ...run, + ...heartbeat.decorateActiveRunStatus(run), outputSilence: await heartbeat.buildRunOutputSilence(run), })))); return; } res.json(await Promise.all(liveRuns.map(async (run) => ({ - ...run, + ...heartbeat.decorateActiveRunStatus(run), outputSilence: await heartbeat.buildRunOutputSilence(run), })))); }); @@ -3538,9 +3538,10 @@ export function agentRoutes( } assertCompanyAccess(req, run.companyId); const retryExhaustedReason = await heartbeat.getRetryExhaustedReason(runId); + const decoratedRun = heartbeat.decorateActiveRunStatus(run); res.json( redactCurrentUserValue( - { ...run, retryExhaustedReason, outputSilence: await heartbeat.buildRunOutputSilence(run) }, + { ...decoratedRun, retryExhaustedReason, outputSilence: await heartbeat.buildRunOutputSilence(run) }, await getCurrentUserRedactionOptions(), ), ); @@ -3732,7 +3733,7 @@ export function agentRoutes( .orderBy(desc(heartbeatRuns.createdAt)); res.json(await Promise.all(liveRuns.map(async (run) => ({ - ...run, + ...heartbeat.decorateActiveRunStatus(run, { companyId: issue.companyId, issueId: issue.id }), outputSilence: await heartbeat.buildRunOutputSilence({ ...run, companyId: issue.companyId }), })))); }); @@ -3777,8 +3778,9 @@ export function agentRoutes( return; } + const decoratedRun = heartbeat.decorateActiveRunStatus(run, { companyId: issue.companyId, issueId: issue.id }); res.json({ - ...run, + ...decoratedRun, agentId: agent.id, agentName: agent.name, adapterType: agent.adapterType, diff --git a/server/src/services/heartbeat-run-runtime-status.test.ts b/server/src/services/heartbeat-run-runtime-status.test.ts new file mode 100644 index 0000000000..457ce76804 --- /dev/null +++ b/server/src/services/heartbeat-run-runtime-status.test.ts @@ -0,0 +1,89 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { + clearAllHeartbeatRunRuntimeStatuses, + clearHeartbeatRunRuntimeStatus, + getHeartbeatRunRuntimeStatus, + MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS, + setHeartbeatRunRuntimeStatus, + sweepExpiredHeartbeatRunRuntimeStatuses, +} from "./heartbeat-run-runtime-status.js"; + +describe("heartbeat run runtime status store", () => { + afterEach(() => { + clearAllHeartbeatRunRuntimeStatuses(); + }); + + it("stores scoped ephemeral status and expires stale entries", () => { + const updatedAt = new Date("2026-06-24T00:00:00.000Z"); + const status = setHeartbeatRunRuntimeStatus({ + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + runId: "run-1", + phase: "config_sync", + message: `Syncing workspace with apiKey: "sk-test-secret" ${"x".repeat(300)}`, + updatedAt, + }); + + expect(status?.message).toContain("***REDACTED***"); + expect(status?.message.length).toBeLessThanOrEqual(MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS); + expect(getHeartbeatRunRuntimeStatus("run-1", { + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + now: new Date("2026-06-24T00:00:30.000Z"), + })).toMatchObject({ + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + runId: "run-1", + phase: "config_sync", + }); + expect(getHeartbeatRunRuntimeStatus("run-1", { companyId: "other-company" })).toBeNull(); + expect(getHeartbeatRunRuntimeStatus("run-1", { + companyId: "company-1", + now: new Date("2026-06-24T00:02:00.001Z"), + })).toBeNull(); + expect(getHeartbeatRunRuntimeStatus("run-1")).toBeNull(); + }); + + it("clears status explicitly", () => { + setHeartbeatRunRuntimeStatus({ + companyId: "company-1", + issueId: null, + agentId: "agent-1", + runId: "run-1", + phase: "finalize", + message: "Finalizing sandbox workspace", + }); + + expect(clearHeartbeatRunRuntimeStatus("run-1")).toBe(true); + expect(getHeartbeatRunRuntimeStatus("run-1")).toBeNull(); + }); + + it("sweeps expired statuses without touching fresh entries", () => { + setHeartbeatRunRuntimeStatus({ + companyId: "company-1", + issueId: null, + agentId: "agent-1", + runId: "stale-run", + phase: "git_sync", + message: "Syncing stale workspace", + updatedAt: new Date("2026-06-24T00:00:00.000Z"), + }); + setHeartbeatRunRuntimeStatus({ + companyId: "company-1", + issueId: null, + agentId: "agent-1", + runId: "fresh-run", + phase: "git_sync", + message: "Syncing fresh workspace", + updatedAt: new Date("2026-06-24T00:01:00.000Z"), + }); + + const now = new Date("2026-06-24T00:01:31.000Z"); + expect(sweepExpiredHeartbeatRunRuntimeStatuses(now)).toBe(1); + expect(getHeartbeatRunRuntimeStatus("stale-run")).toBeNull(); + expect(getHeartbeatRunRuntimeStatus("fresh-run", { now })).toMatchObject({ runId: "fresh-run" }); + }); +}); diff --git a/server/src/services/heartbeat-run-runtime-status.ts b/server/src/services/heartbeat-run-runtime-status.ts new file mode 100644 index 0000000000..fd198751f4 --- /dev/null +++ b/server/src/services/heartbeat-run-runtime-status.ts @@ -0,0 +1,108 @@ +import type { HeartbeatRunStatusPhase } from "@paperclipai/shared"; +import { redactSensitiveText } from "../redaction.js"; + +export const HEARTBEAT_RUN_RUNTIME_STATUS_TTL_MS = 90_000; +export const MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS = 180; + +export interface HeartbeatRunRuntimeStatus { + companyId: string; + issueId: string | null; + agentId: string; + runId: string; + phase: HeartbeatRunStatusPhase; + message: string; + updatedAt: Date; +} + +const runtimeStatusesByRunId = new Map(); + +function cloneStatus(status: HeartbeatRunRuntimeStatus): HeartbeatRunRuntimeStatus { + return { + ...status, + updatedAt: new Date(status.updatedAt), + }; +} + +export function sanitizeHeartbeatRunRuntimeStatusMessage(message: string): string { + const normalized = message.replace(/\s+/g, " ").trim(); + const redacted = redactSensitiveText(normalized); + if (redacted.length <= MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS) return redacted; + return `${redacted.slice(0, MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS - 3)}...`; +} + +function isExpired(status: HeartbeatRunRuntimeStatus, now: Date, ttlMs: number) { + return now.getTime() - status.updatedAt.getTime() > ttlMs; +} + +export function setHeartbeatRunRuntimeStatus( + input: Omit & { + message: string; + updatedAt?: Date; + }, +): HeartbeatRunRuntimeStatus | null { + const message = sanitizeHeartbeatRunRuntimeStatusMessage(input.message); + if (!message) { + clearHeartbeatRunRuntimeStatus(input.runId); + return null; + } + + const status: HeartbeatRunRuntimeStatus = { + companyId: input.companyId, + issueId: input.issueId, + agentId: input.agentId, + runId: input.runId, + phase: input.phase, + message, + updatedAt: input.updatedAt ? new Date(input.updatedAt) : new Date(), + }; + runtimeStatusesByRunId.set(status.runId, status); + return cloneStatus(status); +} + +export function getHeartbeatRunRuntimeStatus( + runId: string, + expected?: { + companyId?: string | null; + issueId?: string | null; + agentId?: string | null; + now?: Date; + ttlMs?: number; + }, +): HeartbeatRunRuntimeStatus | null { + const status = runtimeStatusesByRunId.get(runId); + if (!status) return null; + + const now = expected?.now ?? new Date(); + const ttlMs = expected?.ttlMs ?? HEARTBEAT_RUN_RUNTIME_STATUS_TTL_MS; + if (isExpired(status, now, ttlMs)) { + runtimeStatusesByRunId.delete(runId); + return null; + } + + if (expected?.companyId !== undefined && status.companyId !== expected.companyId) return null; + if (expected?.issueId !== undefined && status.issueId !== expected.issueId) return null; + if (expected?.agentId !== undefined && status.agentId !== expected.agentId) return null; + + return cloneStatus(status); +} + +export function clearHeartbeatRunRuntimeStatus(runId: string): boolean { + return runtimeStatusesByRunId.delete(runId); +} + +export function clearAllHeartbeatRunRuntimeStatuses(): void { + runtimeStatusesByRunId.clear(); +} + +export function sweepExpiredHeartbeatRunRuntimeStatuses( + now = new Date(), + ttlMs = HEARTBEAT_RUN_RUNTIME_STATUS_TTL_MS, +): number { + let swept = 0; + for (const [runId, status] of runtimeStatusesByRunId) { + if (!isExpired(status, now, ttlMs)) continue; + runtimeStatusesByRunId.delete(runId); + swept += 1; + } + return swept; +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index ce89bc9652..197b11d8c6 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -15,6 +15,7 @@ import { type EnvironmentLeaseStatus, type ExecutionWorkspace, type ExecutionWorkspaceConfig, + type HeartbeatRunStatusPhase, type IssueExecutionMonitorClearReason, type IssueExecutionMonitorPolicy, type IssueExecutionMonitorRecoveryPolicy, @@ -181,6 +182,7 @@ import { redactEventPayload, redactSensitiveText } from "../redaction.js"; import { hasSessionCompactionThresholds, resolveSessionCompactionPolicy, + type RuntimeStatusUpdate, type SessionCompactionPolicy, } from "@paperclipai/adapter-utils"; import { @@ -194,6 +196,12 @@ import { environmentRuntimeService } from "./environment-runtime.js"; import { skillVersionSelectionMap } from "./runtime-skill-selections.js"; import { environmentRunOrchestrator } from "./environment-run-orchestrator.js"; import { isUnsafeSessionWorkspaceCwd } from "./session-workspace-cwd.js"; +import { + clearHeartbeatRunRuntimeStatus, + getHeartbeatRunRuntimeStatus, + setHeartbeatRunRuntimeStatus, + sweepExpiredHeartbeatRunRuntimeStatuses, +} from "./heartbeat-run-runtime-status.js"; import { assertLowTrustRuntimeServicesAllowed, assertLowTrustWorkspaceIsolation, @@ -3035,6 +3043,95 @@ function isHeartbeatRunTerminalStatus( ); } +function isHeartbeatRunRuntimeStatusActive(status: string | null | undefined): boolean { + return status === "queued" || status === "running"; +} + +type HeartbeatRunRuntimeStatusRunLike = { + id: string; + status?: string | null; + companyId?: string | null; + agentId?: string | null; + issueId?: string | null; + contextSnapshot?: Record | null; +}; + +function readRuntimeStatusIssueIdCandidate( + run: HeartbeatRunRuntimeStatusRunLike, +): string | null | undefined { + if ("issueId" in run) return readNonEmptyString(run.issueId) ?? null; + if ("contextSnapshot" in run) { + return readNonEmptyString(parseObject(run.contextSnapshot).issueId) ?? null; + } + return undefined; +} + +function decorateHeartbeatRunRuntimeStatus( + run: T, + expected: { + companyId?: string | null; + issueId?: string | null; + agentId?: string | null; + } = {}, +): T & { + currentStatusMessage: string | null; + currentStatusUpdatedAt: Date | null; +} { + if (isHeartbeatRunTerminalStatus(run.status)) { + clearHeartbeatRunRuntimeStatus(run.id); + } + + const companyId = expected.companyId ?? run.companyId ?? null; + const agentId = expected.agentId ?? run.agentId ?? null; + const issueId = + expected.issueId !== undefined ? expected.issueId : readRuntimeStatusIssueIdCandidate(run); + const currentStatus = + isHeartbeatRunRuntimeStatusActive(run.status) && companyId && agentId + ? getHeartbeatRunRuntimeStatus(run.id, { + companyId, + agentId, + ...(issueId !== undefined ? { issueId } : {}), + }) + : null; + + return { + ...run, + currentStatusMessage: currentStatus?.message ?? null, + currentStatusUpdatedAt: currentStatus?.updatedAt ?? null, + }; +} + +function recordHeartbeatRunRuntimeProgress( + run: Pick, + update: RuntimeStatusUpdate, + issueId: string | null, +) { + if (!isHeartbeatRunRuntimeStatusActive(run.status)) return null; + const status = setHeartbeatRunRuntimeStatus({ + companyId: run.companyId, + issueId, + agentId: run.agentId, + runId: run.id, + phase: update.phase as HeartbeatRunStatusPhase, + message: update.message, + }); + if (!status) return null; + + publishLiveEvent({ + companyId: status.companyId, + type: "heartbeat.run.progress", + payload: { + runId: status.runId, + agentId: status.agentId, + issueId: status.issueId, + phase: status.phase, + message: status.message, + updatedAt: status.updatedAt.toISOString(), + }, + }); + return status; +} + export function buildPaperclipTaskMarkdown(input: { issue: { id: string; @@ -3525,6 +3622,25 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? null); } + async function recordCurrentHeartbeatRunRuntimeProgress( + run: Pick, + update: RuntimeStatusUpdate, + issueId: string | null, + ) { + if (!isHeartbeatRunRuntimeStatusActive(run.status)) { + clearHeartbeatRunRuntimeStatus(run.id); + return null; + } + + const currentRun = await getRun(run.id); + if (!currentRun || !isHeartbeatRunRuntimeStatusActive(currentRun.status)) { + clearHeartbeatRunRuntimeStatus(run.id); + return null; + } + + return recordHeartbeatRunRuntimeProgress(currentRun, update, issueId); + } + async function getRunLogAccess(runId: string) { return db .select(heartbeatRunLogAccessColumns) @@ -4912,6 +5028,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? null); if (updated) { + if (isHeartbeatRunTerminalStatus(updated.status)) { + clearHeartbeatRunRuntimeStatus(updated.id); + } publishLiveEvent({ companyId: updated.companyId, type: "heartbeat.run.status", @@ -4946,6 +5065,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? null); if (updated) { + if (isHeartbeatRunTerminalStatus(updated.status)) { + clearHeartbeatRunRuntimeStatus(updated.id); + } publishLiveEvent({ companyId: updated.companyId, type: "heartbeat.run.status", @@ -9568,6 +9690,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) : undefined, onLog, onMeta: onAdapterMeta, + onRuntimeProgress: async (progress) => { + await recordCurrentHeartbeatRunRuntimeProgress(run, progress, issueId); + }, onSpawn: async (meta) => { await persistRunProcessMetadata(run.id, { pid: meta.pid, @@ -12047,6 +12172,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) getRun, + decorateActiveRunStatus: decorateHeartbeatRunRuntimeStatus, + recordRuntimeProgress: recordCurrentHeartbeatRunRuntimeProgress, + sweepExpiredRuntimeStatuses: sweepExpiredHeartbeatRunRuntimeStatuses, + getRunLogAccess, getRuntimeState: async (agentId: string) => { diff --git a/ui/src/context/LiveUpdatesProvider.test.ts b/ui/src/context/LiveUpdatesProvider.test.ts index f696b0b93d..1625fbfc68 100644 --- a/ui/src/context/LiveUpdatesProvider.test.ts +++ b/ui/src/context/LiveUpdatesProvider.test.ts @@ -100,6 +100,49 @@ describe("LiveUpdatesProvider issue invalidation", () => { }); }); + it("keeps heartbeat progress invalidation scoped to live run data", () => { + const invalidations: unknown[] = []; + const queryClient = { + invalidateQueries: (input: unknown) => { + invalidations.push(input); + }, + }; + + __liveUpdatesTestUtils.invalidateHeartbeatProgressQueries( + queryClient as never, + "company-1", + { + agentId: "agent-1", + runId: "run-1", + }, + ); + + expect(invalidations).toContainEqual({ + queryKey: queryKeys.liveRuns("company-1"), + }); + expect(invalidations).toContainEqual({ + queryKey: queryKeys.heartbeats("company-1"), + }); + expect(invalidations).toContainEqual({ + queryKey: queryKeys.agents.list("company-1"), + }); + expect(invalidations).toContainEqual({ + queryKey: queryKeys.agents.detail("agent-1"), + }); + expect(invalidations).toContainEqual({ + queryKey: queryKeys.heartbeats("company-1", "agent-1"), + }); + expect(invalidations).not.toContainEqual({ + queryKey: queryKeys.dashboard("company-1"), + }); + expect(invalidations).not.toContainEqual({ + queryKey: queryKeys.costs("company-1"), + }); + expect(invalidations).not.toContainEqual({ + queryKey: queryKeys.sidebarBadges("company-1"), + }); + }); + it("refreshes issue document caches when a document activity event arrives", () => { const invalidations: unknown[] = []; const queryClient = { diff --git a/ui/src/context/LiveUpdatesProvider.tsx b/ui/src/context/LiveUpdatesProvider.tsx index 086a8d4790..74d66d8708 100644 --- a/ui/src/context/LiveUpdatesProvider.tsx +++ b/ui/src/context/LiveUpdatesProvider.tsx @@ -650,6 +650,22 @@ function invalidateHeartbeatQueries( } } +function invalidateHeartbeatProgressQueries( + queryClient: ReturnType, + companyId: string, + payload: Record, +) { + queryClient.invalidateQueries({ queryKey: queryKeys.liveRuns(companyId) }); + queryClient.invalidateQueries({ queryKey: queryKeys.heartbeats(companyId) }); + queryClient.invalidateQueries({ queryKey: queryKeys.agents.list(companyId) }); + + const agentId = readString(payload.agentId); + if (agentId) { + queryClient.invalidateQueries({ queryKey: queryKeys.agents.detail(agentId) }); + queryClient.invalidateQueries({ queryKey: queryKeys.heartbeats(companyId, agentId) }); + } +} + function invalidateActivityQueries( queryClient: ReturnType, companyId: string, @@ -858,7 +874,10 @@ function handleLiveEvent( return; } - if (event.type === "heartbeat.run.queued" || event.type === "heartbeat.run.status") { + if ( + event.type === "heartbeat.run.queued" || + event.type === "heartbeat.run.status" + ) { invalidateHeartbeatQueries(queryClient, expectedCompanyId, payload); invalidateVisibleIssueRunQueries(queryClient, pathname, payload); if (event.type === "heartbeat.run.status") { @@ -873,6 +892,12 @@ function handleLiveEvent( return; } + if (event.type === "heartbeat.run.progress") { + invalidateHeartbeatProgressQueries(queryClient, expectedCompanyId, payload); + invalidateVisibleIssueRunQueries(queryClient, pathname, payload); + return; + } + if (event.type === "heartbeat.run.event") { return; } @@ -956,6 +981,7 @@ export const __liveUpdatesTestUtils = { closeSocketQuietly, hydrateVisibleIssueComment, invalidateActivityQueries, + invalidateHeartbeatProgressQueries, invalidateVisibleIssueRunQueries, resolveLiveCompanyId, shouldDeferIssueRefetchForVisibleAgentActivity,