Add ephemeral sandbox runtime status plumbing (#8593)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Sandboxed agent runs can spend meaningful time preparing a remote workspace before the agent transcript shows useful output. > - Operators need short, current progress text for those setup phases, but that text should not become durable run history. > - The existing live-run websocket path already carries run updates to the UI, so the backend can reuse that channel instead of adding polling. > - This pull request adds an ephemeral runtime-progress contract, a process-local status store, and heartbeat integration for sandbox-managed runs. > - The benefit is a clearer active-run experience without database migrations or persistent progress rows. ## Linked Issues or Issue Description Refs #248 No exact public GitHub issue was found for this status-message plumbing. The underlying problem is that active sandboxed runs currently have setup phases, such as workspace sync and restore, where the operator cannot see concise current progress through the live run state. This PR addresses that gap for the backend/runtime layer while keeping progress messages ephemeral. GitHub search performed for related or duplicate work: `sandbox runtime status`, `sandbox restore index`, and `runtime progress`. No direct duplicate PR was found. ## What Changed - Added shared runtime-progress types and the `heartbeat.run.progress` live event type. - Added a process-local heartbeat run runtime-status store with TTL, bounded/redacted messages, and terminal cleanup. - Threaded runtime progress callbacks through heartbeat execution and active/live run serialization. - Emitted sandbox-managed runtime phase updates for sync, adapter startup, restore/export, and finalization paths. - Added backend and adapter-utils tests for ephemeral status behavior, terminal cleanup, live serialization, and sandbox progress callbacks. ## Verification - `pnpm install --frozen-lockfile` - Local PII scan before push: high-confidence secret patterns, internal issue links, local user paths, and private URL patterns checked across all three split diffs; no real secrets or internal links found. The only secret-like text is an intentional fake test fixture (`sk-test-secret`). - `git diff --check origin/master..feat/sandbox-runtime-status` - `pnpm exec vitest run server/src/services/heartbeat-run-runtime-status.test.ts server/src/__tests__/heartbeat-runtime-state.test.ts server/src/__tests__/agent-live-run-routes.test.ts packages/adapter-utils/src/sandbox-managed-runtime.test.ts` — 4 files, 23 tests passed. - `pnpm run typecheck` passed on both top stacks that include this branch: `feat/sandbox-status-ui` and `fix/sandbox-restore-index-sync`. - `pnpm run build` passed on both top stacks that include this branch; Vite reported existing CSS `::highlight` and chunk-size warnings. - `pnpm run test:run` was attempted on `fix/sandbox-restore-index-sync`; it failed in two unrelated broad-suite tests. One depends on this host's Git default branch behavior, and one depends on local Claude model-discovery environment. The changed focused suites above pass. ## Risks - Runtime progress is process-local by design, so status disappears after TTL, terminal cleanup, or server restart. - Clients that do not consume `heartbeat.run.progress` simply keep existing behavior. - Message redaction is intentionally generic; overly specific phase details should stay out of runtime-progress payloads. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI GPT-5 via Codex coding agent, with shell/tool execution in a local worktree. Exact context-window metadata is not exposed by the runtime. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip CTO <cto@paperclip.local> Co-authored-by: Paperclip CTO <noreply@paperclip.ing>
This commit is contained in:
Devin Foley
GitHub
parent
bac15ebd09
commit
a27e5ad002
|
|
@ -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<PreparedSandboxManagedRuntime> {
|
||||
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,
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
onRuntimeProgress?: RuntimeStatusSink;
|
||||
onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise<void>;
|
||||
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<PreparedAdapterExecutionTargetRuntime> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
|
||||
export interface RuntimeProgressReporterOptions {
|
||||
sink: RuntimeProgressSink;
|
||||
phase: RuntimeProgressPhase;
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
if (!sink) return;
|
||||
await Promise.resolve(sink({ phase, message })).catch(() => undefined);
|
||||
}
|
||||
|
||||
function mergeExcludes(...groups: Array<string[] | undefined>): 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<PreparedSandboxManagedRuntime> {
|
||||
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 });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
onMeta?: (meta: AdapterInvocationMeta) => Promise<void>;
|
||||
onRuntimeProgress?: RuntimeStatusSink;
|
||||
onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise<void>;
|
||||
authToken?: string;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -487,6 +487,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
assets: [
|
||||
{
|
||||
key: "skills",
|
||||
|
|
@ -790,6 +791,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
timeoutSec,
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
terminalResultCleanup: {
|
||||
graceMs: terminalResultCleanupGraceMs,
|
||||
|
|
|
|||
|
|
@ -464,6 +464,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
assets: [
|
||||
{
|
||||
key: "home",
|
||||
|
|
@ -875,6 +876,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
timeoutSec,
|
||||
graceSec,
|
||||
onSpawn: wrappedOnSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: async (stream, chunk) => {
|
||||
if (stream === "stdout") {
|
||||
monitor?.noteStdoutChunk(chunk);
|
||||
|
|
|
|||
|
|
@ -365,6 +365,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
assets: [{
|
||||
key: "skills",
|
||||
localDir: localSkillsDir,
|
||||
|
|
@ -636,6 +637,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
graceSec,
|
||||
stdin: prompt,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: async (stream, chunk) => {
|
||||
if (stream !== "stdout") {
|
||||
await onLog(stream, chunk);
|
||||
|
|
|
|||
|
|
@ -341,6 +341,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
assets: [{
|
||||
key: "skills",
|
||||
localDir: localSkillsDir,
|
||||
|
|
@ -591,6 +592,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
timeoutSec,
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
});
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -326,6 +326,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
installCommand: ctx.runtimeCommandSpec?.installCommand ?? null,
|
||||
detectCommand: ctx.runtimeCommandSpec?.detectCommand ?? command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
});
|
||||
restoreRemoteWorkspace = () =>
|
||||
preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
|
|
@ -474,6 +475,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
timeoutSec,
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
});
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -378,6 +378,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
assets: [
|
||||
{
|
||||
key: "skills",
|
||||
|
|
@ -603,6 +604,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
timeoutSec,
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
});
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -418,6 +418,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
assets: [
|
||||
{
|
||||
key: "skills",
|
||||
|
|
@ -709,6 +710,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
timeoutSec,
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: bufferedOnLog,
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -720,6 +720,7 @@ export type RunLivenessState = (typeof RUN_LIVENESS_STATES)[number];
|
|||
export const LIVE_EVENT_TYPES = [
|
||||
"heartbeat.run.queued",
|
||||
"heartbeat.run.status",
|
||||
"heartbeat.run.progress",
|
||||
"heartbeat.run.event",
|
||||
"heartbeat.run.log",
|
||||
"agent.status",
|
||||
|
|
|
|||
|
|
@ -661,6 +661,7 @@ export type {
|
|||
AgentWakeupSkipped,
|
||||
HeartbeatRun,
|
||||
HeartbeatRunEvent,
|
||||
HeartbeatRunStatusPhase,
|
||||
AgentRuntimeState,
|
||||
AgentTaskSession,
|
||||
AgentWakeupRequest,
|
||||
|
|
|
|||
|
|
@ -56,8 +56,29 @@ export interface HeartbeatRun {
|
|||
createdAt: Date;
|
||||
updatedAt: Date;
|
||||
outputSilence?: HeartbeatRunOutputSilence;
|
||||
/**
|
||||
* Ephemeral, process-local current status message for an active run. Resolved
|
||||
* from the in-memory runtime status store (never persisted to the database)
|
||||
* and only populated for active/live run reads. Disappears on TTL expiry,
|
||||
* terminal run status, or server restart.
|
||||
*/
|
||||
currentStatusMessage?: string | null;
|
||||
currentStatusUpdatedAt?: Date | string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Typed phase labels emitted by the sandbox-managed runtime as it progresses
|
||||
* through workspace preparation, adapter startup, restore/export, and
|
||||
* finalization. Used by the ephemeral runtime status plumbing; not persisted.
|
||||
*/
|
||||
export type HeartbeatRunStatusPhase =
|
||||
| "git_sync"
|
||||
| "config_sync"
|
||||
| "adapter_startup"
|
||||
| "restore"
|
||||
| "export"
|
||||
| "finalize";
|
||||
|
||||
export type HeartbeatRunOutputSilenceLevel =
|
||||
| "not_applicable"
|
||||
| "ok"
|
||||
|
|
|
|||
|
|
@ -452,6 +452,7 @@ export type {
|
|||
AgentWakeupSkipped,
|
||||
HeartbeatRun,
|
||||
HeartbeatRunEvent,
|
||||
HeartbeatRunStatusPhase,
|
||||
AgentRuntimeState,
|
||||
AgentTaskSession,
|
||||
AgentWakeupRequest,
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ const mockAgentService = vi.hoisted(() => ({
|
|||
|
||||
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(),
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -852,6 +852,14 @@ export async function startServer(): Promise<StartedServer> {
|
|||
});
|
||||
|
||||
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) => {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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" });
|
||||
});
|
||||
});
|
||||
|
|
@ -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<string, HeartbeatRunRuntimeStatus>();
|
||||
|
||||
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<HeartbeatRunRuntimeStatus, "message" | "updatedAt"> & {
|
||||
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;
|
||||
}
|
||||
|
|
@ -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<string, unknown> | 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<T extends HeartbeatRunRuntimeStatusRunLike>(
|
||||
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<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "status" | "contextSnapshot">,
|
||||
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<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "status" | "contextSnapshot">,
|
||||
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) => {
|
||||
|
|
|
|||
|
|
@ -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 = {
|
||||
|
|
|
|||
|
|
@ -650,6 +650,22 @@ function invalidateHeartbeatQueries(
|
|||
}
|
||||
}
|
||||
|
||||
function invalidateHeartbeatProgressQueries(
|
||||
queryClient: ReturnType<typeof useQueryClient>,
|
||||
companyId: string,
|
||||
payload: Record<string, unknown>,
|
||||
) {
|
||||
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<typeof useQueryClient>,
|
||||
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,
|
||||
|
|
|
|||
Loading…
Reference in New Issue