Improve live agent feedback during sandboxed runs (#8915)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A core part of that experience is watching active agent runs without dropping into raw logs first > - Local and sandbox-backed adapters already record useful run output, progress, and tool activity > - But active issue threads could sit visually stale while the agent was syncing workspaces, tailing sandbox output, or emitting incremental tool-call updates > - Operators need timely, human-readable progress while preserving the raw transcript underneath > - This pull request streams sandbox run-log progress into runtime status, keeps visible issue threads refreshed, and folds repeated ACPX tool updates into stable transcript cards > - The benefit is that long-running agent work becomes easier to supervise without changing the task/comment control-plane model ## Linked Issues or Issue Description No public GitHub issue exists for this exact change. Problem/motivation: - During long-running sandboxed agent work, the issue UI can appear idle even though the agent is actively syncing, running tools, or producing incremental output. - Operators need realtime feedback at the issue-thread layer, not only after opening raw logs or waiting for the final heartbeat result. - Related public context: #1808 previously added live-run status dots to Projects; #4362 touches heartbeat wakeup behavior but is not a duplicate of this runtime/UI feedback change. ## What Changed - Added sandbox run-log streaming support and defaulted sandbox-capable local adapters into the richer live-feedback path. - Surfaced environment/sandbox sync progress through heartbeat runtime status with bounded, redacted snippets. - Added live issue-thread cache patching so visible active runs update as progress events arrive. - Folded repeated ACPX `tool_call` updates into one transcript card instead of stacking duplicate cards. - Updated adapter docs and added focused regression coverage for sandbox log streaming, runtime status, ACPX parsing, live updates, transcript rendering, and issue chat messages. ## Verification - `pnpm install --frozen-lockfile` - `pnpm exec vitest run ui/src/context/LiveUpdatesProvider.test.ts` - `pnpm exec vitest run server/src/services/heartbeat-run-runtime-status.test.ts server/src/__tests__/heartbeat-runtime-state.test.ts ui/src/context/LiveUpdatesProvider.test.ts` - `pnpm exec vitest run packages/adapter-utils/src/execution-target-sandbox.test.ts packages/adapter-utils/src/sandbox-managed-runtime.test.ts server/src/services/heartbeat-run-runtime-status.test.ts server/src/__tests__/agent-live-run-routes.test.ts server/src/__tests__/heartbeat-runtime-state.test.ts packages/adapters/acpx-local/src/ui/parse-stdout.test.ts ui/src/context/LiveUpdatesProvider.test.ts ui/src/components/transcript/RunTranscriptView.test.tsx ui/src/lib/issue-chat-messages.test.ts ui/src/components/IssueChatThread.test.tsx` - GitHub PR workflow on head `8397953e7b41ccd42e5d9457ee7e4dfb996e4ec5`: `verify`, build, typecheck/release-registry, e2e, general shards, serialized server shards, and canary dry run passed. - Greptile Review on head `8397953e7b41ccd42e5d9457ee7e4dfb996e4ec5`: Confidence Score 5/5, no unresolved review threads. ## Risks - Live issue-thread cache patching could miss an edge case for a route shape not covered by tests. - Surfacing active-run snippets needs continued care around redaction; this PR keeps snippets bounded and adds redaction-focused coverage. - More frequent active-run UI refreshes could expose performance issues on very large issue threads, though updates are scoped to visible run/query caches. ## Model Used OpenAI GPT-5 via Codex, operating as a tool-enabled coding agent with shell, git, and repository-editing capabilities. Context window size is not exposed in this 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 <noreply@paperclip.ing>
This commit is contained in:
Devin Foley
GitHub
parent
bf982c8c83
commit
c48feee190
|
|
@ -20,6 +20,7 @@ When a heartbeat fires, Paperclip:
|
|||
|---------|----------|-------------|
|
||||
| [Claude Code](/adapters/claude-local) | `claude_local` | Runs Claude Code CLI locally |
|
||||
| [Codex](/adapters/codex-local) | `codex_local` | Runs OpenAI Codex CLI locally |
|
||||
| ACPX Local | `acpx_local` | Runs Claude, Codex, or a custom ACP agent through ACPX with live structured event streaming |
|
||||
| [Gemini CLI](/adapters/gemini-local) | `gemini_local` | Runs Gemini CLI locally (experimental — adapter package exists, not yet in stable type enum) |
|
||||
| OpenCode | `opencode_local` | Runs OpenCode CLI locally (multi-provider `provider/model`) |
|
||||
| Cursor | `cursor` | Runs Cursor in background mode |
|
||||
|
|
@ -92,12 +93,25 @@ my-adapter/
|
|||
|
||||
## Choosing an Adapter
|
||||
|
||||
- **Need a coding agent?** Use `claude_local`, `codex_local`, `opencode_local`, `hermes_local`, or install `droid_local` as an external plugin
|
||||
- **Need a coding agent?** Use `claude_local`, `codex_local`, `acpx_local`, `opencode_local`, `hermes_local`, or install `droid_local` as an external plugin
|
||||
- **Need the richest live run feedback (especially for sandbox workers)?** Use `acpx_local` — see [Feedback granularity](#feedback-granularity)
|
||||
- **Need Hermes on another host or already running as a service?** Use `hermes_gateway`
|
||||
- **Need to run a script or command?** Use `process`
|
||||
- **Need to call a custom external service?** Use `http`
|
||||
- **Need something custom?** [Create your own adapter](/adapters/creating-an-adapter) or [build an external adapter plugin](/adapters/external-adapters)
|
||||
|
||||
## Feedback Granularity
|
||||
|
||||
Adapter choice determines how much structured, live detail a run's transcript can show while the agent is still working. Every adapter's stdout is streamed to the run log and rendered live in the UI — including runs on sandbox execution targets, whose logs are tailed and delivered incrementally — but the *granularity* of what you see depends on the event stream the adapter emits.
|
||||
|
||||
Rough tiers, richest first:
|
||||
|
||||
1. **`acpx_local` — full structured event stream.** ACPX emits a JSONL event per meaningful runtime moment: `acpx.session` (agent, mode, session identity), `acpx.status` (progress text plus context-window usage), `acpx.text_delta` (assistant/thinking token deltas), `acpx.tool_call` (tool title, call id, and status updates as the call progresses), `acpx.result` (stop reason summary), and `acpx.error` (code, message, retryability). The transcript renders these as live-updating message, thinking, tool, and status blocks, and repeated `acpx.tool_call` status updates fold into a single tool card instead of stacking duplicates.
|
||||
2. **CLI wrappers (`claude_local`, `codex_local`, `cursor`, `opencode_local`, …).** These parse each CLI's own streaming JSON output. You get assistant text, tool calls/results, and a final usage/cost summary, but granularity is limited to what the CLI prints — some emit tool progress, others only call/finish pairs.
|
||||
3. **Generic adapters (`process`, `http`).** Plain stdout/stderr lines with no structured transcript — you see raw output only.
|
||||
|
||||
**Recommendation:** for sandbox workers, prefer `acpx_local`. Sandbox run logs are streamed live, so the richer the event stream, the more useful the live transcript and status line are while a remote run is in flight. ACPX's status events (including context usage) and incremental tool-call updates give the closest thing to watching the agent work locally.
|
||||
|
||||
## UI Parser Contract
|
||||
|
||||
External adapters can ship a self-contained UI parser that tells the Paperclip web UI how to render their stdout. Without it, the UI uses a generic shell parser. See the [UI Parser Contract](/adapters/adapter-ui-parser) for details.
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import {
|
|||
startAdapterExecutionTargetPaperclipBridge,
|
||||
type AdapterSandboxExecutionTarget,
|
||||
} from "./execution-target.js";
|
||||
import { createSandboxRunLogTailFactory } from "./sandbox-run-log-stream.js";
|
||||
import { runChildProcess } from "./server-utils.js";
|
||||
import { shellQuote } from "./ssh.js";
|
||||
|
||||
|
|
@ -75,6 +76,33 @@ describe("sandbox adapter execution targets", () => {
|
|||
return contents;
|
||||
}
|
||||
|
||||
function encodeTailTick(stdout: Buffer, stderr: Buffer): string {
|
||||
return [
|
||||
"__PAPERCLIP_RUN_LOG_STDOUT__",
|
||||
stdout.toString("base64"),
|
||||
"__PAPERCLIP_RUN_LOG_STDERR__",
|
||||
stderr.toString("base64"),
|
||||
"__PAPERCLIP_RUN_LOG_END__",
|
||||
"",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
async function waitForCondition(predicate: () => boolean, message: string): Promise<void> {
|
||||
const deadline = Date.now() + 1000;
|
||||
while (Date.now() < deadline) {
|
||||
if (predicate()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
}
|
||||
throw new Error(message);
|
||||
}
|
||||
|
||||
function combinedStream(
|
||||
events: Array<{ stream: "stdout" | "stderr"; chunk: string }>,
|
||||
stream: "stdout" | "stderr",
|
||||
): string {
|
||||
return events.filter((event) => event.stream === stream).map((event) => event.chunk).join("");
|
||||
}
|
||||
|
||||
it("executes through the provider-neutral runner without a remote spec", async () => {
|
||||
const runner = {
|
||||
execute: vi.fn(async () => ({
|
||||
|
|
@ -464,6 +492,245 @@ describe("sandbox adapter execution targets", () => {
|
|||
}
|
||||
});
|
||||
|
||||
it("creates a sandbox run log tail factory when bridge streaming is enabled", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-stream-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remoteCwd = path.join(rootDir, "workspace");
|
||||
const runtimeRootDir = path.join(remoteCwd, ".paperclip-runtime", "codex");
|
||||
await mkdir(runtimeRootDir, { recursive: true });
|
||||
|
||||
const logs: Array<{ stream: "stdout" | "stderr"; chunk: string }> = [];
|
||||
const target: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "e2b",
|
||||
environmentId: "env-1",
|
||||
leaseId: "lease-1",
|
||||
remoteCwd,
|
||||
runner: createLocalSandboxRunner(),
|
||||
streamRunLogs: true,
|
||||
timeoutMs: 30_000,
|
||||
};
|
||||
|
||||
const bridge = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-bridge-stream",
|
||||
target,
|
||||
runtimeRootDir,
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: "http://127.0.0.1:9",
|
||||
onLog: async (stream, chunk) => {
|
||||
logs.push({ stream, chunk });
|
||||
},
|
||||
});
|
||||
try {
|
||||
expect(bridge?.runLogTail).toBeTruthy();
|
||||
expect(combinedStream(logs, "stdout")).toContain("Sandbox run log streaming enabled");
|
||||
|
||||
const wrapped = bridge!.runLogTail!.create().wrapCommand("agent-cli", ["--message", "hello world"]);
|
||||
expect(wrapped.command).toBe("sh");
|
||||
expect(wrapped.args.join("\n")).toContain("tee -a");
|
||||
expect(wrapped.args.join("\n")).toContain("agent-cli");
|
||||
} finally {
|
||||
await bridge?.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("defaults sandbox run log streaming on and honors the explicit opt-out", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-stream-default-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remoteCwd = path.join(rootDir, "workspace");
|
||||
const runtimeRootDir = path.join(remoteCwd, ".paperclip-runtime", "codex");
|
||||
await mkdir(runtimeRootDir, { recursive: true });
|
||||
|
||||
const baseTarget: AdapterSandboxExecutionTarget = {
|
||||
kind: "remote",
|
||||
transport: "sandbox",
|
||||
providerKey: "e2b",
|
||||
environmentId: "env-1",
|
||||
leaseId: "lease-1",
|
||||
remoteCwd,
|
||||
runner: createLocalSandboxRunner(),
|
||||
timeoutMs: 30_000,
|
||||
};
|
||||
|
||||
const defaultBridge = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-bridge-stream-default",
|
||||
target: baseTarget,
|
||||
runtimeRootDir,
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: "http://127.0.0.1:9",
|
||||
});
|
||||
try {
|
||||
expect(defaultBridge?.runLogTail).toBeTruthy();
|
||||
} finally {
|
||||
await defaultBridge?.stop();
|
||||
}
|
||||
|
||||
const optOutBridge = await startAdapterExecutionTargetPaperclipBridge({
|
||||
runId: "run-bridge-stream-opt-out",
|
||||
target: { ...baseTarget, streamRunLogs: false },
|
||||
runtimeRootDir,
|
||||
adapterKey: "codex",
|
||||
hostApiToken: "real-run-jwt",
|
||||
hostApiUrl: "http://127.0.0.1:9",
|
||||
});
|
||||
try {
|
||||
expect(optOutBridge?.runLogTail ?? null).toBeNull();
|
||||
} finally {
|
||||
await optOutBridge?.stop();
|
||||
}
|
||||
});
|
||||
|
||||
it("tails sandbox run log chunks with byte offsets and dedupes the final batch", async () => {
|
||||
const stdoutText = "stdout-abc\n";
|
||||
const stderrText = "stderr-xyz\n";
|
||||
const stdoutBytes = Buffer.from(stdoutText, "utf8");
|
||||
const stderrBytes = Buffer.from(stderrText, "utf8");
|
||||
const stdoutOffsets: number[] = [];
|
||||
const stderrOffsets: number[] = [];
|
||||
const events: Array<{ stream: "stdout" | "stderr"; chunk: string }> = [];
|
||||
|
||||
const runner = {
|
||||
execute: vi.fn(async (input: {
|
||||
command: string;
|
||||
args?: string[];
|
||||
cwd?: string;
|
||||
env?: Record<string, string>;
|
||||
timeoutMs?: number;
|
||||
}) => {
|
||||
const script = input.args?.[1] ?? "";
|
||||
const offsets = [...script.matchAll(/tail -c \+(\d+) /g)].map((match) => Number(match[1]));
|
||||
const stdoutStart = Math.max(0, (offsets[0] ?? 1) - 1);
|
||||
const stderrStart = Math.max(0, (offsets[1] ?? 1) - 1);
|
||||
stdoutOffsets.push(stdoutStart + 1);
|
||||
stderrOffsets.push(stderrStart + 1);
|
||||
return {
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
stdout: encodeTailTick(
|
||||
stdoutBytes.subarray(stdoutStart, stdoutStart + 4),
|
||||
stderrBytes.subarray(stderrStart, stderrStart + 4),
|
||||
),
|
||||
stderr: "",
|
||||
pid: null,
|
||||
startedAt: new Date().toISOString(),
|
||||
};
|
||||
}),
|
||||
};
|
||||
|
||||
const tail = createSandboxRunLogTailFactory({
|
||||
runner,
|
||||
remoteCwd: "/workspace",
|
||||
logsDir: "/workspace/.paperclip-runtime/codex/paperclip-bridge/queue/logs",
|
||||
pollIntervalMs: 1,
|
||||
maxChunkBytesPerTick: 4,
|
||||
tickTimeoutMs: 50,
|
||||
}).create();
|
||||
|
||||
tail.start(async (stream, chunk) => {
|
||||
events.push({ stream, chunk });
|
||||
});
|
||||
|
||||
await waitForCondition(
|
||||
() => combinedStream(events, "stdout") === stdoutText && combinedStream(events, "stderr") === stderrText,
|
||||
"run log tail did not stream expected stdout/stderr chunks",
|
||||
);
|
||||
|
||||
await tail.finish({ stdout: stdoutText, stderr: stderrText });
|
||||
|
||||
expect(combinedStream(events, "stdout")).toBe(stdoutText);
|
||||
expect(combinedStream(events, "stderr")).toBe(stderrText);
|
||||
expect(stdoutOffsets.slice(0, 3)).toEqual([1, 5, 9]);
|
||||
expect(stderrOffsets.slice(0, 3)).toEqual([1, 5, 9]);
|
||||
expect(runner.execute).toHaveBeenCalledWith(expect.objectContaining({
|
||||
command: "sh",
|
||||
cwd: "/workspace",
|
||||
env: { PAPERCLIP_SANDBOX_EXEC_CHANNEL: "bridge" },
|
||||
timeoutMs: 50,
|
||||
}));
|
||||
});
|
||||
|
||||
it("emits only the unstreamed final suffix when the tail loop stops early", async () => {
|
||||
const finalStdout = "prefix suffix\n";
|
||||
const finalBytes = Buffer.from(finalStdout, "utf8");
|
||||
const events: Array<{ stream: "stdout" | "stderr"; chunk: string }> = [];
|
||||
|
||||
const runner = {
|
||||
execute: vi.fn(async (input: { args?: string[] }) => {
|
||||
const script = input.args?.[1] ?? "";
|
||||
const offsets = [...script.matchAll(/tail -c \+(\d+) /g)].map((match) => Number(match[1]));
|
||||
const stdoutStart = Math.max(0, (offsets[0] ?? 1) - 1);
|
||||
return {
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
stdout: encodeTailTick(finalBytes.subarray(stdoutStart, stdoutStart + 7), Buffer.alloc(0)),
|
||||
stderr: "",
|
||||
pid: null,
|
||||
startedAt: new Date().toISOString(),
|
||||
};
|
||||
}),
|
||||
};
|
||||
|
||||
const tail = createSandboxRunLogTailFactory({
|
||||
runner,
|
||||
remoteCwd: "/workspace",
|
||||
logsDir: "/workspace/.paperclip-runtime/codex/paperclip-bridge/queue/logs",
|
||||
pollIntervalMs: 1,
|
||||
maxChunkBytesPerTick: 7,
|
||||
tickTimeoutMs: 50,
|
||||
}).create();
|
||||
|
||||
tail.start(async (stream, chunk) => {
|
||||
events.push({ stream, chunk });
|
||||
});
|
||||
await waitForCondition(() => combinedStream(events, "stdout").length >= 7, "run log tail did not emit prefix");
|
||||
await tail.finish({ stdout: finalStdout, stderr: "" });
|
||||
|
||||
expect(combinedStream(events, "stdout")).toBe(finalStdout);
|
||||
expect(events.filter((event) => event.stream === "stdout").map((event) => event.chunk).join("|"))
|
||||
.toBe("prefix |suffix\n");
|
||||
});
|
||||
|
||||
it("delivers the final batch and a warning when run log polling degrades", async () => {
|
||||
const events: Array<{ stream: "stdout" | "stderr"; chunk: string }> = [];
|
||||
const runner = {
|
||||
execute: vi.fn(async () => ({
|
||||
exitCode: 1,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
stdout: "",
|
||||
stderr: "tail failed",
|
||||
pid: null,
|
||||
startedAt: new Date().toISOString(),
|
||||
})),
|
||||
};
|
||||
|
||||
const tail = createSandboxRunLogTailFactory({
|
||||
runner,
|
||||
remoteCwd: "/workspace",
|
||||
logsDir: "/workspace/.paperclip-runtime/codex/paperclip-bridge/queue/logs",
|
||||
pollIntervalMs: 1,
|
||||
tickTimeoutMs: 50,
|
||||
maxConsecutiveFailures: 1,
|
||||
}).create();
|
||||
|
||||
tail.start(async (stream, chunk) => {
|
||||
events.push({ stream, chunk });
|
||||
});
|
||||
await waitForCondition(() => runner.execute.mock.calls.length >= 1, "run log tail did not poll before finish");
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
await tail.finish({ stdout: "final out\n", stderr: "final err\n" });
|
||||
|
||||
expect(combinedStream(events, "stdout")).toBe("final out\n");
|
||||
expect(combinedStream(events, "stderr")).toBe(
|
||||
"final err\n[paperclip] Run log streaming degraded during the run; remaining output was delivered at completion.\n",
|
||||
);
|
||||
});
|
||||
|
||||
it("exposes the Paperclip bridge to the sandbox shell surface", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-execution-target-bridge-shell-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
|
|
|
|||
|
|
@ -15,9 +15,14 @@ import {
|
|||
createSandboxCallbackBridgeAsset,
|
||||
createSandboxCallbackBridgeToken,
|
||||
DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES,
|
||||
sandboxCallbackBridgeDirectories,
|
||||
startSandboxCallbackBridgeServer,
|
||||
startSandboxCallbackBridgeWorker,
|
||||
} from "./sandbox-callback-bridge.js";
|
||||
import {
|
||||
createSandboxRunLogTailFactory,
|
||||
type SandboxRunLogTailFactory,
|
||||
} from "./sandbox-run-log-stream.js";
|
||||
import { createSshCommandManagedRuntimeRunner, parseSshRemoteExecutionSpec, runSshCommand, shellQuote } from "./ssh.js";
|
||||
import {
|
||||
ensureCommandResolvable,
|
||||
|
|
@ -57,6 +62,13 @@ export interface AdapterSandboxExecutionTarget {
|
|||
remoteCwd: string;
|
||||
timeoutMs?: number | null;
|
||||
runner?: CommandManagedRuntimeRunner;
|
||||
/**
|
||||
* Sandbox-backed adapter runs stream the agent CLI's stdout/stderr
|
||||
* incrementally via a log-tail loop beside the callback bridge instead of
|
||||
* waiting for the batched provider result. Streaming is ON by default;
|
||||
* set to `false` to explicitly opt out back to batch-at-end delivery.
|
||||
*/
|
||||
streamRunLogs?: boolean | null;
|
||||
}
|
||||
|
||||
export type AdapterExecutionTarget =
|
||||
|
|
@ -86,6 +98,12 @@ export interface AdapterExecutionTargetProcessOptions {
|
|||
onRuntimeProgress?: RuntimeStatusSink;
|
||||
onSpawn?: (meta: { pid: number; processGroupId: number | null; startedAt: string }) => Promise<void>;
|
||||
terminalResultCleanup?: TerminalResultCleanupOptions;
|
||||
/**
|
||||
* Sandbox-only: factory from the Paperclip bridge handle that streams the
|
||||
* CLI's stdout/stderr during the run. When provided, the batched provider
|
||||
* onLog is suppressed and incremental chunks flow through `onLog` instead.
|
||||
*/
|
||||
runLogTail?: SandboxRunLogTailFactory | null;
|
||||
}
|
||||
|
||||
export interface AdapterExecutionTargetShellOptions {
|
||||
|
|
@ -98,6 +116,12 @@ export interface AdapterExecutionTargetShellOptions {
|
|||
|
||||
export interface AdapterExecutionTargetPaperclipBridgeHandle {
|
||||
env: Record<string, string>;
|
||||
/**
|
||||
* Present when the sandbox target opted into run-log streaming
|
||||
* (`streamRunLogs`). Create one handle per CLI attempt and pass it to
|
||||
* `runAdapterExecutionTargetProcess` via `options.runLogTail`.
|
||||
*/
|
||||
runLogTail?: SandboxRunLogTailFactory | null;
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
|
|
@ -412,18 +436,38 @@ export async function runAdapterExecutionTargetProcess(
|
|||
phase: "adapter_startup",
|
||||
message: "Starting adapter in sandbox",
|
||||
});
|
||||
return await runner.execute({
|
||||
command,
|
||||
args,
|
||||
cwd: target.remoteCwd,
|
||||
env,
|
||||
stdin: options.stdin,
|
||||
timeoutMs: options.timeoutSec > 0 ? options.timeoutSec * 1000 : target.timeoutMs ?? undefined,
|
||||
onLog: options.onLog,
|
||||
onSpawn: options.onSpawn
|
||||
? async (meta) => options.onSpawn?.({ ...meta, processGroupId: null })
|
||||
: undefined,
|
||||
});
|
||||
const runLogTail = options.runLogTail?.create() ?? null;
|
||||
let execCommand = command;
|
||||
let execArgs = args;
|
||||
if (runLogTail) {
|
||||
({ command: execCommand, args: execArgs } = runLogTail.wrapCommand(command, args));
|
||||
runLogTail.start(options.onLog);
|
||||
}
|
||||
try {
|
||||
const result = await runner.execute({
|
||||
command: execCommand,
|
||||
args: execArgs,
|
||||
cwd: target.remoteCwd,
|
||||
env,
|
||||
stdin: options.stdin,
|
||||
timeoutMs: options.timeoutSec > 0 ? options.timeoutSec * 1000 : target.timeoutMs ?? undefined,
|
||||
// The tail loop already streams incremental chunks; suppress the
|
||||
// runner's end-of-run batched onLog to avoid duplicate log bytes.
|
||||
onLog: runLogTail ? undefined : options.onLog,
|
||||
onSpawn: options.onSpawn
|
||||
? async (meta) => options.onSpawn?.({ ...meta, processGroupId: null })
|
||||
: undefined,
|
||||
});
|
||||
if (runLogTail) {
|
||||
await runLogTail.finish({ stdout: result.stdout, stderr: result.stderr });
|
||||
}
|
||||
return result;
|
||||
} catch (error) {
|
||||
if (runLogTail) {
|
||||
await runLogTail.abort();
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
const env =
|
||||
|
|
@ -886,6 +930,7 @@ export function parseAdapterExecutionTarget(value: unknown): AdapterExecutionTar
|
|||
leaseId: readStringMeta(parsed, "leaseId"),
|
||||
remoteCwd,
|
||||
timeoutMs: typeof parsed.timeoutMs === "number" ? parsed.timeoutMs : null,
|
||||
streamRunLogs: typeof parsed.streamRunLogs === "boolean" ? parsed.streamRunLogs : null,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -1194,12 +1239,25 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
|
|||
throw error;
|
||||
}
|
||||
|
||||
let runLogTail: SandboxRunLogTailFactory | null = null;
|
||||
if (target.transport === "sandbox" && target.streamRunLogs !== false) {
|
||||
runLogTail = createSandboxRunLogTailFactory({
|
||||
runner,
|
||||
remoteCwd: target.remoteCwd,
|
||||
logsDir: sandboxCallbackBridgeDirectories(queueDir).logsDir,
|
||||
shellCommand,
|
||||
});
|
||||
await onLog("stdout", "[paperclip] Sandbox run log streaming enabled for this run.\n");
|
||||
}
|
||||
|
||||
return {
|
||||
env: {
|
||||
PAPERCLIP_API_URL: server.baseUrl,
|
||||
PAPERCLIP_API_KEY: bridgeToken,
|
||||
PAPERCLIP_API_BRIDGE_MODE: "queue_v1",
|
||||
PAPERCLIP_BRIDGE_QUEUE_DIR: queueDir,
|
||||
},
|
||||
runLogTail,
|
||||
stop: async () => {
|
||||
await Promise.allSettled([
|
||||
server?.stop(),
|
||||
|
|
|
|||
|
|
@ -32,6 +32,9 @@ export type RuntimeStatusPhase =
|
|||
export interface RuntimeStatusUpdate {
|
||||
phase: RuntimeStatusPhase;
|
||||
message: string;
|
||||
currentToolName?: string | null;
|
||||
lastAssistantSnippet?: string | null;
|
||||
lastEventAt?: Date | string | null;
|
||||
}
|
||||
|
||||
export type RuntimeStatusSink = (update: RuntimeStatusUpdate) => void | Promise<void>;
|
||||
|
|
|
|||
|
|
@ -148,12 +148,18 @@ 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([
|
||||
expect(runtimeStatuses).toEqual(expect.arrayContaining([
|
||||
"config_sync:Syncing workspace to sandbox",
|
||||
"config_sync:Syncing runtime assets to sandbox",
|
||||
"restore:Restoring workspace from sandbox",
|
||||
"finalize:Finalizing sandbox workspace",
|
||||
]);
|
||||
]));
|
||||
expect(runtimeStatuses).toEqual(expect.arrayContaining([
|
||||
expect.stringMatching(/^config_sync:Syncing workspace to sandbox: 100% \(\d+\.\d\/\d+\.\d MB\)$/),
|
||||
expect.stringMatching(/^config_sync:Syncing skills to sandbox: 100% \(\d+\.\d\/\d+\.\d MB\)$/),
|
||||
expect.stringMatching(/^restore:Restoring workspace from sandbox: 100% \(\d+\.\d\/\d+\.\d MB\)$/),
|
||||
]));
|
||||
expect(runtimeStatuses.at(-1)).toBe("finalize:Finalizing sandbox workspace");
|
||||
});
|
||||
|
||||
it("syncs git-backed workspaces through a shallow standalone clone and keeps .git out of archives", async () => {
|
||||
|
|
@ -185,19 +191,29 @@ describe("sandbox managed runtime", () => {
|
|||
|
||||
const uploadedTars: { remotePath: string; bytes: Buffer }[] = [];
|
||||
const downloadedTars: { remotePath: string; bytes: Buffer }[] = [];
|
||||
const driveProgress = async (
|
||||
total: number,
|
||||
onProgress: ((done: number, total: number | null) => void | Promise<void>) | undefined,
|
||||
) => {
|
||||
if (!onProgress) return;
|
||||
await onProgress(Math.max(1, Math.floor(total / 2)), total);
|
||||
await onProgress(total, total);
|
||||
};
|
||||
const client: SandboxManagedRuntimeClient = {
|
||||
makeDir: async (remotePath) => {
|
||||
await mkdir(remotePath, { recursive: true });
|
||||
},
|
||||
writeFile: async (remotePath, bytes) => {
|
||||
writeFile: async (remotePath, bytes, options) => {
|
||||
await mkdir(path.dirname(remotePath), { recursive: true });
|
||||
const buffer = Buffer.from(bytes);
|
||||
if (remotePath.endsWith("-upload.tar")) uploadedTars.push({ remotePath, bytes: buffer });
|
||||
await writeFile(remotePath, buffer);
|
||||
await driveProgress(buffer.byteLength, options?.onProgress);
|
||||
},
|
||||
readFile: async (remotePath) => {
|
||||
readFile: async (remotePath, options) => {
|
||||
const buffer = await readFile(remotePath);
|
||||
if (remotePath.endsWith("workspace-download.tar")) downloadedTars.push({ remotePath, bytes: buffer });
|
||||
await driveProgress(buffer.byteLength, options?.onProgress);
|
||||
return buffer;
|
||||
},
|
||||
listFiles: async () => [],
|
||||
|
|
@ -208,7 +224,7 @@ describe("sandbox managed runtime", () => {
|
|||
await execFile("sh", ["-c", command], { maxBuffer: 32 * 1024 * 1024 });
|
||||
},
|
||||
};
|
||||
const runtimeStatusPhases: string[] = [];
|
||||
const runtimeStatuses: Array<{ phase: string; message: string }> = [];
|
||||
|
||||
const prepared = await prepareSandboxManagedRuntime({
|
||||
spec: {
|
||||
|
|
@ -223,7 +239,7 @@ describe("sandbox managed runtime", () => {
|
|||
client,
|
||||
workspaceLocalDir: localWorkspaceDir,
|
||||
onRuntimeProgress: async (status) => {
|
||||
runtimeStatusPhases.push(status.phase);
|
||||
runtimeStatuses.push({ phase: status.phase, message: status.message });
|
||||
},
|
||||
});
|
||||
|
||||
|
|
@ -269,13 +285,21 @@ 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([
|
||||
expect(runtimeStatuses.map((status) => status.phase)).toEqual(expect.arrayContaining([
|
||||
"git_sync",
|
||||
"config_sync",
|
||||
"export",
|
||||
"restore",
|
||||
"finalize",
|
||||
]);
|
||||
]));
|
||||
expect(runtimeStatuses.some((status) => (
|
||||
status.phase === "git_sync" &&
|
||||
/^Syncing git history to sandbox: 100% \(\d+\.\d\/\d+\.\d MB\)$/.test(status.message)
|
||||
))).toBe(true);
|
||||
expect(runtimeStatuses.some((status) => (
|
||||
status.phase === "export" &&
|
||||
/^Exporting git history from sandbox: 100% \(\d+\.\d\/\d+\.\d MB\)$/.test(status.message)
|
||||
))).toBe(true);
|
||||
});
|
||||
|
||||
it("repairs stale host index deletions when the sandbox restores a clean git worktree", async () => {
|
||||
|
|
|
|||
|
|
@ -382,10 +382,28 @@ function makeTransferProgress(
|
|||
phase: RuntimeProgressPhase,
|
||||
direction: RuntimeProgressDirection,
|
||||
label?: string,
|
||||
): { options: SandboxTransferProgressOptions | undefined; finish: () => Promise<void> } {
|
||||
if (!sink) return { options: undefined, finish: async () => {} };
|
||||
runtimeStatus?: {
|
||||
sink: RuntimeStatusSink | undefined;
|
||||
phase: RuntimeStatusPhase;
|
||||
},
|
||||
): {
|
||||
options: SandboxTransferProgressOptions | undefined;
|
||||
finish: (doneBytes?: number, totalBytes?: number | null) => Promise<void>;
|
||||
} {
|
||||
if (!sink && !runtimeStatus?.sink) {
|
||||
return { options: undefined, finish: async () => {} };
|
||||
}
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink,
|
||||
sink: async (line) => {
|
||||
await sink?.(line);
|
||||
if (runtimeStatus?.sink) {
|
||||
await emitRuntimeStatus(
|
||||
runtimeStatus.sink,
|
||||
runtimeStatus.phase,
|
||||
line.replace(/^\[paperclip\]\s*/, "").trim(),
|
||||
);
|
||||
}
|
||||
},
|
||||
phase,
|
||||
direction,
|
||||
target: "sandbox",
|
||||
|
|
@ -397,8 +415,8 @@ function makeTransferProgress(
|
|||
await reporter.report(transferredBytes, totalBytes);
|
||||
},
|
||||
},
|
||||
finish: async () => {
|
||||
await reporter.complete();
|
||||
finish: async (doneBytes, totalBytes) => {
|
||||
await reporter.complete(doneBytes, totalBytes);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -460,9 +478,15 @@ export async function prepareSandboxManagedRuntime(input: {
|
|||
const gitTarBytes = await fs.readFile(gitTarPath);
|
||||
const remoteGitTar = path.posix.join(runtimeRootDir, "git-workspace-upload.tar");
|
||||
await input.client.makeDir(runtimeRootDir);
|
||||
const gitUpload = makeTransferProgress(input.onProgress, "Syncing", "to", "git history");
|
||||
const gitUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
"git history",
|
||||
{ sink: input.onRuntimeProgress, phase: "git_sync" },
|
||||
);
|
||||
await input.client.writeFile(remoteGitTar, toArrayBuffer(gitTarBytes), gitUpload.options);
|
||||
await gitUpload.finish();
|
||||
await gitUpload.finish(gitTarBytes.byteLength, gitTarBytes.byteLength);
|
||||
await input.client.run(
|
||||
`sh -c ${shellQuote(
|
||||
`mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
|
|
@ -494,13 +518,19 @@ export async function prepareSandboxManagedRuntime(input: {
|
|||
const workspaceTarBytes = await fs.readFile(workspaceTarPath);
|
||||
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-upload.tar");
|
||||
await input.client.makeDir(runtimeRootDir);
|
||||
const workspaceUpload = makeTransferProgress(input.onProgress, "Syncing", "to", "workspace");
|
||||
const workspaceUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
"workspace",
|
||||
{ sink: input.onRuntimeProgress, phase: "config_sync" },
|
||||
);
|
||||
await input.client.writeFile(
|
||||
remoteWorkspaceTar,
|
||||
toArrayBuffer(workspaceTarBytes),
|
||||
workspaceUpload.options,
|
||||
);
|
||||
await workspaceUpload.finish();
|
||||
await workspaceUpload.finish(workspaceTarBytes.byteLength, workspaceTarBytes.byteLength);
|
||||
const extractWorkspaceTarCommand = gitSnapshot
|
||||
? `mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`tar -xf ${shellQuote(remoteWorkspaceTar)} -C ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
|
|
@ -534,9 +564,15 @@ export async function prepareSandboxManagedRuntime(input: {
|
|||
const assetTarBytes = await fs.readFile(assetTarPath);
|
||||
const remoteAssetDir = path.posix.join(runtimeRootDir, asset.key);
|
||||
const remoteAssetTar = path.posix.join(runtimeRootDir, `${asset.key}-upload.tar`);
|
||||
const assetUpload = makeTransferProgress(input.onProgress, "Syncing", "to", asset.key);
|
||||
const assetUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
asset.key,
|
||||
{ sink: input.onRuntimeProgress, phase: "config_sync" },
|
||||
);
|
||||
await input.client.writeFile(remoteAssetTar, toArrayBuffer(assetTarBytes), assetUpload.options);
|
||||
await assetUpload.finish();
|
||||
await assetUpload.finish(assetTarBytes.byteLength, assetTarBytes.byteLength);
|
||||
await input.client.run(
|
||||
`sh -c ${shellQuote(
|
||||
`rm -rf ${shellQuote(remoteAssetDir)} && ` +
|
||||
|
|
@ -582,9 +618,16 @@ export async function prepareSandboxManagedRuntime(input: {
|
|||
}))}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
);
|
||||
const gitExport = makeTransferProgress(restoreSink, "Exporting git history", "from");
|
||||
const gitExport = makeTransferProgress(
|
||||
restoreSink,
|
||||
"Exporting git history",
|
||||
"from",
|
||||
undefined,
|
||||
{ sink: input.onRuntimeProgress, phase: "export" },
|
||||
);
|
||||
const bundleBytes = await input.client.readFile(remoteGitBundle, gitExport.options);
|
||||
await gitExport.finish();
|
||||
const bundleBuffer = toBuffer(bundleBytes);
|
||||
await gitExport.finish(bundleBuffer.byteLength, bundleBuffer.byteLength);
|
||||
await input.client.remove(remoteGitBundle).catch(() => undefined);
|
||||
remoteWorkspaceStatus = await input.client.readFile(remoteWorkspaceStatusPath)
|
||||
.then((bytes) => toBuffer(bytes).toString("utf8").trim())
|
||||
|
|
@ -592,7 +635,7 @@ export async function prepareSandboxManagedRuntime(input: {
|
|||
remoteWorkspaceStatus = remoteWorkspaceStatus === "clean" ? "clean" : "dirty";
|
||||
await input.client.remove(remoteWorkspaceStatusPath).catch(() => undefined);
|
||||
const bundlePath = path.join(tempDir, "git-delta.bundle");
|
||||
await fs.writeFile(bundlePath, toBuffer(bundleBytes));
|
||||
await fs.writeFile(bundlePath, bundleBuffer);
|
||||
importedHead = await fetchGitBundleIntoLocalRef({
|
||||
localDir: input.workspaceLocalDir,
|
||||
bundlePath,
|
||||
|
|
@ -612,13 +655,20 @@ export async function prepareSandboxManagedRuntime(input: {
|
|||
}))}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
);
|
||||
const workspaceRestore = makeTransferProgress(restoreSink, "Restoring", "from", "workspace");
|
||||
const workspaceRestore = makeTransferProgress(
|
||||
restoreSink,
|
||||
"Restoring",
|
||||
"from",
|
||||
"workspace",
|
||||
{ sink: input.onRuntimeProgress, phase: "restore" },
|
||||
);
|
||||
const archiveBytes = await input.client.readFile(remoteWorkspaceTar, workspaceRestore.options);
|
||||
await workspaceRestore.finish();
|
||||
const archiveBuffer = toBuffer(archiveBytes);
|
||||
await workspaceRestore.finish(archiveBuffer.byteLength, archiveBuffer.byteLength);
|
||||
await input.client.remove(remoteWorkspaceTar).catch(() => undefined);
|
||||
const localArchivePath = path.join(tempDir, "workspace.tar");
|
||||
const extractedDir = path.join(tempDir, "workspace");
|
||||
await fs.writeFile(localArchivePath, toBuffer(archiveBytes));
|
||||
await fs.writeFile(localArchivePath, archiveBuffer);
|
||||
await extractTarballToDirectory({
|
||||
archivePath: localArchivePath,
|
||||
localDir: extractedDir,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,278 @@
|
|||
import path from "node:path";
|
||||
import { StringDecoder } from "node:string_decoder";
|
||||
import type { CommandManagedRuntimeRunner } from "./command-managed-runtime.js";
|
||||
import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js";
|
||||
import { shellQuote } from "./ssh.js";
|
||||
|
||||
// Sandbox providers execute commands through batch RPCs, so agent CLI output
|
||||
// normally only reaches the host when the process exits. This module streams
|
||||
// that output during the run instead: `wrapCommand` tees the CLI's
|
||||
// stdout/stderr into log files under the bridge runtime directory inside the
|
||||
// sandbox, and a host-side poll loop tails those files (byte offsets +
|
||||
// base64 transport, mirroring the callback-bridge queue client) and emits
|
||||
// incremental `onLog` chunks through the existing run-log pipeline.
|
||||
|
||||
const SANDBOX_EXEC_CHANNEL_ENV = "PAPERCLIP_SANDBOX_EXEC_CHANNEL";
|
||||
const SANDBOX_EXEC_CHANNEL_BRIDGE = "bridge";
|
||||
|
||||
const DEFAULT_TAIL_POLL_INTERVAL_MS = 250;
|
||||
const DEFAULT_TAIL_MAX_CHUNK_BYTES = 64 * 1024;
|
||||
const DEFAULT_TAIL_TICK_TIMEOUT_MS = 15_000;
|
||||
const DEFAULT_TAIL_MAX_CONSECUTIVE_FAILURES = 3;
|
||||
|
||||
const TAIL_MARKER_STDOUT = "__PAPERCLIP_RUN_LOG_STDOUT__";
|
||||
const TAIL_MARKER_STDERR = "__PAPERCLIP_RUN_LOG_STDERR__";
|
||||
const TAIL_MARKER_END = "__PAPERCLIP_RUN_LOG_END__";
|
||||
|
||||
export type SandboxRunLogSink = (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
|
||||
export interface SandboxRunLogTailHandle {
|
||||
/**
|
||||
* Wrap the agent CLI invocation in a shell script that tees stdout/stderr
|
||||
* into tailable log files while preserving the original streams (the
|
||||
* provider result must keep the full stdout for adapter parsing) and the
|
||||
* original exit code.
|
||||
*/
|
||||
wrapCommand(command: string, args: string[]): { command: string; args: string[] };
|
||||
/** Start the host-side poll loop that tails the log files via the runner. */
|
||||
start(onLog: SandboxRunLogSink): void;
|
||||
/**
|
||||
* Stop the poll loop and emit any bytes of the final batched output that
|
||||
* were not already streamed. Emitting the suffix past the streamed byte
|
||||
* offset both dedupes the final batch and guarantees full coverage when
|
||||
* the tail loop degraded mid-run.
|
||||
*/
|
||||
finish(finalBatch: { stdout: string; stderr: string }): Promise<void>;
|
||||
/** Stop the poll loop without emitting anything further (error path). */
|
||||
abort(): Promise<void>;
|
||||
}
|
||||
|
||||
export interface SandboxRunLogTailFactory {
|
||||
create(): SandboxRunLogTailHandle;
|
||||
}
|
||||
|
||||
export interface SandboxRunLogTailFactoryOptions {
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
remoteCwd: string;
|
||||
/** Remote directory the log files live in (bridge queue `logs/` dir). */
|
||||
logsDir: string;
|
||||
shellCommand?: "bash" | "sh" | null;
|
||||
pollIntervalMs?: number | null;
|
||||
maxChunkBytesPerTick?: number | null;
|
||||
tickTimeoutMs?: number | null;
|
||||
maxConsecutiveFailures?: number | null;
|
||||
}
|
||||
|
||||
function normalizePositiveInt(value: number | null | undefined, fallback: number): number {
|
||||
return typeof value === "number" && Number.isFinite(value) && value > 0
|
||||
? Math.trunc(value)
|
||||
: fallback;
|
||||
}
|
||||
|
||||
interface TailStreamState {
|
||||
stream: "stdout" | "stderr";
|
||||
logFile: string;
|
||||
offset: number;
|
||||
decoder: StringDecoder;
|
||||
}
|
||||
|
||||
function decodeBase64Section(lines: string[]): Buffer {
|
||||
const joined = lines.join("").replace(/\s+/g, "");
|
||||
if (joined.length === 0) return Buffer.alloc(0);
|
||||
return Buffer.from(joined, "base64");
|
||||
}
|
||||
|
||||
export function createSandboxRunLogTailFactory(
|
||||
options: SandboxRunLogTailFactoryOptions,
|
||||
): SandboxRunLogTailFactory {
|
||||
const shellCommand = preferredShellForSandbox(options.shellCommand);
|
||||
const pollIntervalMs = normalizePositiveInt(options.pollIntervalMs, DEFAULT_TAIL_POLL_INTERVAL_MS);
|
||||
const maxChunkBytes = normalizePositiveInt(options.maxChunkBytesPerTick, DEFAULT_TAIL_MAX_CHUNK_BYTES);
|
||||
const tickTimeoutMs = normalizePositiveInt(options.tickTimeoutMs, DEFAULT_TAIL_TICK_TIMEOUT_MS);
|
||||
const maxConsecutiveFailures = normalizePositiveInt(
|
||||
options.maxConsecutiveFailures,
|
||||
DEFAULT_TAIL_MAX_CONSECUTIVE_FAILURES,
|
||||
);
|
||||
|
||||
let sequence = 0;
|
||||
|
||||
function createHandle(): SandboxRunLogTailHandle {
|
||||
sequence += 1;
|
||||
const baseName = `run-${sequence}`;
|
||||
const stdoutLog = path.posix.join(options.logsDir, `${baseName}-stdout.log`);
|
||||
const stderrLog = path.posix.join(options.logsDir, `${baseName}-stderr.log`);
|
||||
const statusFile = path.posix.join(options.logsDir, `${baseName}-status`);
|
||||
|
||||
const streams: [TailStreamState, TailStreamState] = [
|
||||
{ stream: "stdout", logFile: stdoutLog, offset: 0, decoder: new StringDecoder("utf8") },
|
||||
{ stream: "stderr", logFile: stderrLog, offset: 0, decoder: new StringDecoder("utf8") },
|
||||
];
|
||||
|
||||
let sink: SandboxRunLogSink | null = null;
|
||||
let stopped = false;
|
||||
let degraded = false;
|
||||
let loopPromise: Promise<void> | null = null;
|
||||
let wakeSleep: (() => void) | null = null;
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const timer = setTimeout(() => {
|
||||
wakeSleep = null;
|
||||
resolve();
|
||||
}, ms);
|
||||
wakeSleep = () => {
|
||||
clearTimeout(timer);
|
||||
wakeSleep = null;
|
||||
resolve();
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
function buildTickScript(): string {
|
||||
const lines: string[] = [`printf '%s\\n' ${shellQuote(TAIL_MARKER_STDOUT)}`];
|
||||
for (const state of streams) {
|
||||
if (state.stream === "stderr") {
|
||||
lines.push(`printf '%s\\n' ${shellQuote(TAIL_MARKER_STDERR)}`);
|
||||
}
|
||||
lines.push(
|
||||
`if [ -f ${shellQuote(state.logFile)} ]; then tail -c +${state.offset + 1} ${shellQuote(state.logFile)} | head -c ${maxChunkBytes} | base64; fi`,
|
||||
);
|
||||
}
|
||||
lines.push(`printf '%s\\n' ${shellQuote(TAIL_MARKER_END)}`);
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
function parseTickOutput(stdout: string): { stdout: Buffer; stderr: Buffer } | null {
|
||||
const lines = stdout.split(/\r?\n/);
|
||||
const stdoutIndex = lines.indexOf(TAIL_MARKER_STDOUT);
|
||||
const stderrIndex = lines.indexOf(TAIL_MARKER_STDERR);
|
||||
const endIndex = lines.indexOf(TAIL_MARKER_END);
|
||||
if (stdoutIndex < 0 || stderrIndex < stdoutIndex || endIndex < stderrIndex) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
stdout: decodeBase64Section(lines.slice(stdoutIndex + 1, stderrIndex)),
|
||||
stderr: decodeBase64Section(lines.slice(stderrIndex + 1, endIndex)),
|
||||
};
|
||||
}
|
||||
|
||||
async function emitBytes(state: TailStreamState, bytes: Buffer): Promise<void> {
|
||||
if (bytes.length === 0) return;
|
||||
state.offset += bytes.length;
|
||||
const text = state.decoder.write(bytes);
|
||||
if (text.length > 0 && sink) {
|
||||
await sink(state.stream, text);
|
||||
}
|
||||
}
|
||||
|
||||
async function tick(): Promise<void> {
|
||||
const result = await options.runner.execute({
|
||||
command: shellCommand,
|
||||
args: shellCommandArgs(buildTickScript()),
|
||||
cwd: options.remoteCwd,
|
||||
env: { [SANDBOX_EXEC_CHANNEL_ENV]: SANDBOX_EXEC_CHANNEL_BRIDGE },
|
||||
timeoutMs: tickTimeoutMs,
|
||||
});
|
||||
if (result.timedOut || (result.exitCode ?? 1) !== 0) {
|
||||
throw new Error(
|
||||
`Run log tail tick failed (exit ${result.exitCode ?? "null"}${result.timedOut ? ", timed out" : ""}).`,
|
||||
);
|
||||
}
|
||||
const sections = parseTickOutput(result.stdout);
|
||||
if (!sections) {
|
||||
throw new Error("Run log tail tick returned unparseable output.");
|
||||
}
|
||||
await emitBytes(streams[0], sections.stdout);
|
||||
await emitBytes(streams[1], sections.stderr);
|
||||
}
|
||||
|
||||
async function loop(): Promise<void> {
|
||||
let consecutiveFailures = 0;
|
||||
while (!stopped) {
|
||||
await sleep(pollIntervalMs);
|
||||
if (stopped) break;
|
||||
try {
|
||||
await tick();
|
||||
consecutiveFailures = 0;
|
||||
} catch {
|
||||
consecutiveFailures += 1;
|
||||
if (consecutiveFailures >= maxConsecutiveFailures) {
|
||||
degraded = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function stopLoop(): Promise<void> {
|
||||
stopped = true;
|
||||
wakeSleep?.();
|
||||
if (loopPromise) {
|
||||
await loopPromise.catch(() => undefined);
|
||||
loopPromise = null;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
wrapCommand(command, args) {
|
||||
const quotedInvocation = [command, ...args].map(shellQuote).join(" ");
|
||||
// Tee stdout/stderr into tailable log files while keeping both
|
||||
// streams flowing to the provider result. fd 3 carries the real
|
||||
// stdout out of the inner group so stderr can ride the inner pipe
|
||||
// into its own tee. The exit status survives the pipeline through
|
||||
// the status file.
|
||||
const script = [
|
||||
`out_log=${shellQuote(stdoutLog)}`,
|
||||
`err_log=${shellQuote(stderrLog)}`,
|
||||
`status_file=${shellQuote(statusFile)}`,
|
||||
`mkdir -p ${shellQuote(options.logsDir)}`,
|
||||
`: > "$out_log"`,
|
||||
`: > "$err_log"`,
|
||||
`rm -f "$status_file"`,
|
||||
`{`,
|
||||
` { ${quotedInvocation} 3>&-; printf '%s' "$?" > "$status_file"; } 2>&1 1>&3 | tee -a "$err_log" >&2`,
|
||||
`} 3>&1 | tee -a "$out_log"`,
|
||||
`if [ -s "$status_file" ]; then exit "$(cat "$status_file")"; fi`,
|
||||
`exit 1`,
|
||||
].join("\n");
|
||||
return { command: shellCommand, args: shellCommandArgs(script) };
|
||||
},
|
||||
start(onLog) {
|
||||
if (loopPromise || stopped) return;
|
||||
sink = onLog;
|
||||
loopPromise = loop();
|
||||
},
|
||||
async finish(finalBatch) {
|
||||
await stopLoop();
|
||||
if (!sink) return;
|
||||
for (const state of streams) {
|
||||
const finalBytes = Buffer.from(
|
||||
state.stream === "stdout" ? finalBatch.stdout : finalBatch.stderr,
|
||||
"utf8",
|
||||
);
|
||||
if (finalBytes.length > state.offset) {
|
||||
const text = state.decoder.write(finalBytes.subarray(state.offset));
|
||||
if (text.length > 0) {
|
||||
await sink(state.stream, text);
|
||||
}
|
||||
}
|
||||
const rest = state.decoder.end();
|
||||
if (rest.length > 0) {
|
||||
await sink(state.stream, rest);
|
||||
}
|
||||
}
|
||||
if (degraded) {
|
||||
await sink(
|
||||
"stderr",
|
||||
"[paperclip] Run log streaming degraded during the run; remaining output was delivered at completion.\n",
|
||||
);
|
||||
}
|
||||
},
|
||||
async abort() {
|
||||
await stopLoop();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
return { create: createHandle };
|
||||
}
|
||||
|
|
@ -25,10 +25,14 @@ Adapter: acpx_local
|
|||
Use when:
|
||||
- The agent should run through Agent Client Protocol via ACPX on the Paperclip host or a managed execution environment.
|
||||
- You want one built-in adapter that can target Claude, Codex, or a custom ACP server command.
|
||||
- You need Paperclip-managed session identity and live streamed ACP events in later ACPX runtime phases.
|
||||
- You want the richest live run feedback. acpx_local streams structured JSONL events (acpx.session, acpx.status, acpx.text_delta, acpx.tool_call, acpx.result, acpx.error) that the UI renders as live message, thinking, tool, and status blocks.
|
||||
- The agent runs on a sandbox execution target. Sandbox run logs stream live, so
|
||||
acpx_local's granular events make remote runs as observable as local ones.
|
||||
Prefer acpx_local for sandbox workers, especially when watching progress in
|
||||
real time.
|
||||
|
||||
Don't use when:
|
||||
- You need today's stable Claude Code or Codex CLI wrapper behavior. Use claude_local or codex_local until acpx_local runtime execution is enabled.
|
||||
- You depend on CLI-wrapper-specific behavior of claude_local or codex_local (their session files, CLI flags, or CLI-version-specific output).
|
||||
- The host cannot satisfy ACPX's Node >=22.12.0 prerequisite.
|
||||
- The agent runtime is not an ACP server and cannot be launched through ACPX.
|
||||
|
||||
|
|
|
|||
|
|
@ -1096,6 +1096,8 @@ async function emitRuntimeEvent(ctx: AdapterExecutionContext, event: AcpRuntimeE
|
|||
return;
|
||||
}
|
||||
if (event.type === "tool_call") {
|
||||
const eventRecord = event as Record<string, unknown>;
|
||||
const toolInput = eventRecord.input;
|
||||
await emitAcpxLog(ctx, {
|
||||
type: "acpx.tool_call",
|
||||
name: event.title ?? "acp_tool",
|
||||
|
|
@ -1103,6 +1105,7 @@ async function emitRuntimeEvent(ctx: AdapterExecutionContext, event: AcpRuntimeE
|
|||
status: event.status,
|
||||
text: event.text,
|
||||
tag: event.tag,
|
||||
...(toolInput !== undefined ? { input: toolInput } : {}),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -96,6 +96,60 @@ describe("parseAcpxStdoutLine", () => {
|
|||
]);
|
||||
});
|
||||
|
||||
it("merges explicit tool_call input payload with status text", () => {
|
||||
const entries = parseAcpxStdoutLine(
|
||||
emit({
|
||||
type: "acpx.tool_call",
|
||||
name: "read",
|
||||
toolCallId: "tool-3",
|
||||
status: "in_progress",
|
||||
text: "reading README.md",
|
||||
input: { file: "README.md" },
|
||||
}),
|
||||
TS,
|
||||
);
|
||||
expect(entries).toEqual([
|
||||
{
|
||||
kind: "tool_call",
|
||||
ts: TS,
|
||||
name: "read",
|
||||
toolUseId: "tool-3",
|
||||
input: { file: "README.md", status: "in_progress", text: "reading README.md" },
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps terminal tool_call status while preserving existing input", () => {
|
||||
const entries = parseAcpxStdoutLine(
|
||||
emit({
|
||||
type: "acpx.tool_call",
|
||||
name: "read",
|
||||
toolCallId: "tool-4",
|
||||
status: "completed",
|
||||
text: "ok",
|
||||
input: { file: "README.md", status: "running" },
|
||||
}),
|
||||
TS,
|
||||
);
|
||||
expect(entries).toEqual([
|
||||
{
|
||||
kind: "tool_call",
|
||||
ts: TS,
|
||||
name: "read",
|
||||
toolUseId: "tool-4",
|
||||
input: { file: "README.md", status: "running", text: "ok" },
|
||||
},
|
||||
{
|
||||
kind: "tool_result",
|
||||
ts: TS,
|
||||
toolUseId: "tool-4",
|
||||
toolName: "read",
|
||||
content: "ok",
|
||||
isError: false,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it("emits a paired tool_result entry when a tool_call reports terminal status", () => {
|
||||
const completed = parseAcpxStdoutLine(
|
||||
emit({
|
||||
|
|
|
|||
|
|
@ -36,6 +36,11 @@ function pickToolUseId(parsed: Record<string, unknown>): string {
|
|||
);
|
||||
}
|
||||
|
||||
function asRecord(value: unknown): Record<string, unknown> | null {
|
||||
if (typeof value !== "object" || value === null || Array.isArray(value)) return null;
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function statusText(parsed: Record<string, unknown>): string {
|
||||
const text = asString(parsed.text).trim();
|
||||
const tag = asString(parsed.tag).trim();
|
||||
|
|
@ -86,12 +91,19 @@ export function parseAcpxStdoutLine(line: string, ts: string): TranscriptEntry[]
|
|||
const text = asString(parsed.text);
|
||||
const name = asString(parsed.name, "acp_tool");
|
||||
const toolUseId = pickToolUseId(parsed);
|
||||
const input =
|
||||
parsed.input !== undefined
|
||||
? parsed.input
|
||||
: text || status
|
||||
? { ...(text ? { text } : {}), ...(status ? { status } : {}) }
|
||||
: {};
|
||||
const parsedInput = parsed.input;
|
||||
const input = (() => {
|
||||
if (parsedInput === undefined) {
|
||||
return text || status ? { ...(text ? { text } : {}), ...(status ? { status } : {}) } : {};
|
||||
}
|
||||
const inputRecord = asRecord(parsedInput);
|
||||
if (!inputRecord) return parsedInput;
|
||||
return {
|
||||
...inputRecord,
|
||||
...(status && inputRecord.status === undefined ? { status } : {}),
|
||||
...(text && inputRecord.text === undefined ? { text } : {}),
|
||||
};
|
||||
})();
|
||||
const entries: TranscriptEntry[] = [
|
||||
{
|
||||
kind: "tool_call",
|
||||
|
|
|
|||
|
|
@ -793,6 +793,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
terminalResultCleanup: {
|
||||
graceMs: terminalResultCleanupGraceMs,
|
||||
hasTerminalResult: ({ stdout }) => parseClaudeStreamJson(stdout).resultJson !== null,
|
||||
|
|
|
|||
|
|
@ -893,6 +893,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
if (!cleaned.trim()) return;
|
||||
await onLog(stream, cleaned);
|
||||
},
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
});
|
||||
const cleanedStderr = stripCodexRolloutNoise(proc.stderr);
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -645,6 +645,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
}
|
||||
await flushStdoutChunk(chunk);
|
||||
},
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
});
|
||||
await flushStdoutChunk("", true);
|
||||
|
||||
|
|
|
|||
|
|
@ -594,6 +594,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
});
|
||||
return {
|
||||
proc,
|
||||
|
|
|
|||
|
|
@ -606,6 +606,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
});
|
||||
return {
|
||||
proc,
|
||||
|
|
|
|||
|
|
@ -712,6 +712,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: bufferedOnLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
});
|
||||
|
||||
// Flush any remaining buffer content
|
||||
|
|
|
|||
|
|
@ -28,12 +28,16 @@ export interface FakeSandboxEnvironmentConfig {
|
|||
provider: "fake";
|
||||
image: string;
|
||||
reuseLease: boolean;
|
||||
/** Stream agent CLI stdout/stderr during sandbox runs (bridge log-tail loop). */
|
||||
streamRunLogs?: boolean;
|
||||
}
|
||||
|
||||
export interface PluginSandboxEnvironmentConfig {
|
||||
provider: SandboxEnvironmentProvider;
|
||||
reuseLease: boolean;
|
||||
timeoutMs?: number;
|
||||
/** Stream agent CLI stdout/stderr during sandbox runs (bridge log-tail loop). */
|
||||
streamRunLogs?: boolean;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -64,6 +64,9 @@ export interface HeartbeatRun {
|
|||
*/
|
||||
currentStatusMessage?: string | null;
|
||||
currentStatusUpdatedAt?: Date | string | null;
|
||||
currentToolName?: string | null;
|
||||
lastAssistantSnippet?: string | null;
|
||||
lastEventAt?: Date | string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -77,7 +80,8 @@ export type HeartbeatRunStatusPhase =
|
|||
| "adapter_startup"
|
||||
| "restore"
|
||||
| "export"
|
||||
| "finalize";
|
||||
| "finalize"
|
||||
| "run_activity";
|
||||
|
||||
export type HeartbeatRunOutputSilenceLevel =
|
||||
| "not_applicable"
|
||||
|
|
|
|||
|
|
@ -316,6 +316,9 @@ describe("agent live run routes", () => {
|
|||
...run,
|
||||
currentStatusMessage: "Syncing workspace to sandbox",
|
||||
currentStatusUpdatedAt: new Date("2026-04-10T09:30:05.000Z"),
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Inspecting files",
|
||||
lastEventAt: new Date("2026-04-10T09:30:06.000Z"),
|
||||
}));
|
||||
|
||||
const res = await requestApp(
|
||||
|
|
@ -331,6 +334,9 @@ describe("agent live run routes", () => {
|
|||
expect(res.body).toMatchObject({
|
||||
currentStatusMessage: "Syncing workspace to sandbox",
|
||||
currentStatusUpdatedAt: "2026-04-10T09:30:05.000Z",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Inspecting files",
|
||||
lastEventAt: "2026-04-10T09:30:06.000Z",
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -148,6 +148,9 @@ describeEmbeddedPostgres("heartbeat runtime state deduplication", () => {
|
|||
const status = await heartbeat.recordRuntimeProgress(run, {
|
||||
phase: "config_sync",
|
||||
message: "Syncing workspace to sandbox",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Inspecting the repository",
|
||||
lastEventAt: "2026-06-24T00:00:05.000Z",
|
||||
}, issueId);
|
||||
|
||||
expect(status).toMatchObject({
|
||||
|
|
@ -157,6 +160,9 @@ describeEmbeddedPostgres("heartbeat runtime state deduplication", () => {
|
|||
runId,
|
||||
phase: "config_sync",
|
||||
message: "Syncing workspace to sandbox",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Inspecting the repository",
|
||||
lastEventAt: new Date("2026-06-24T00:00:05.000Z"),
|
||||
});
|
||||
expect(heartbeat.decorateActiveRunStatus({
|
||||
id: runId,
|
||||
|
|
@ -166,6 +172,9 @@ describeEmbeddedPostgres("heartbeat runtime state deduplication", () => {
|
|||
status: "running",
|
||||
})).toMatchObject({
|
||||
currentStatusMessage: "Syncing workspace to sandbox",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Inspecting the repository",
|
||||
lastEventAt: new Date("2026-06-24T00:00:05.000Z"),
|
||||
});
|
||||
expect(liveEvents).toContainEqual(expect.objectContaining({
|
||||
companyId,
|
||||
|
|
@ -176,6 +185,9 @@ describeEmbeddedPostgres("heartbeat runtime state deduplication", () => {
|
|||
issueId,
|
||||
phase: "config_sync",
|
||||
message: "Syncing workspace to sandbox",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Inspecting the repository",
|
||||
lastEventAt: "2026-06-24T00:00:05.000Z",
|
||||
}),
|
||||
}));
|
||||
|
||||
|
|
|
|||
|
|
@ -75,6 +75,7 @@ const fakeSandboxEnvironmentConfigSchema = z.object({
|
|||
.min(1, "Fake sandbox environments require an image.")
|
||||
.default("ubuntu:24.04"),
|
||||
reuseLease: z.boolean().optional().default(false),
|
||||
streamRunLogs: z.boolean().optional(),
|
||||
}).strict();
|
||||
|
||||
const pluginSandboxProviderKeySchema = z.string()
|
||||
|
|
@ -89,6 +90,7 @@ const pluginSandboxEnvironmentConfigSchema = z.object({
|
|||
provider: pluginSandboxProviderKeySchema,
|
||||
timeoutMs: z.coerce.number().int().min(1).max(86_400_000).optional(),
|
||||
reuseLease: z.boolean().optional().default(false),
|
||||
streamRunLogs: z.boolean().optional(),
|
||||
}).catchall(z.unknown());
|
||||
|
||||
const pluginEnvironmentConfigSchema = z.object({
|
||||
|
|
|
|||
|
|
@ -73,6 +73,10 @@ export async function resolveEnvironmentExecutionTarget(input: {
|
|||
environmentId: input.environment.id ?? null,
|
||||
leaseId: input.leaseId ?? null,
|
||||
timeoutMs,
|
||||
// Run-log streaming defaults ON for sandbox environments so agent CLI
|
||||
// output reaches the UI mid-run; `streamRunLogs: false` is an explicit
|
||||
// opt-out back to batch-at-end delivery.
|
||||
streamRunLogs: parsed.config.streamRunLogs !== false,
|
||||
runner: input.environmentRuntime && input.lease
|
||||
? {
|
||||
supportsSingleStreamStdinProgress: false,
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import {
|
|||
MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS,
|
||||
setHeartbeatRunRuntimeStatus,
|
||||
sweepExpiredHeartbeatRunRuntimeStatuses,
|
||||
touchHeartbeatRunRuntimeStatus,
|
||||
} from "./heartbeat-run-runtime-status.js";
|
||||
|
||||
describe("heartbeat run runtime status store", () => {
|
||||
|
|
@ -22,11 +23,17 @@ describe("heartbeat run runtime status store", () => {
|
|||
runId: "run-1",
|
||||
phase: "config_sync",
|
||||
message: `Syncing workspace with apiKey: "sk-test-secret" ${"x".repeat(300)}`,
|
||||
currentToolName: `bash apiKey: "sk-tool-secret" ${"x".repeat(120)}`,
|
||||
lastAssistantSnippet: `Reading apiKey: "sk-snippet-secret" ${"x".repeat(300)}`,
|
||||
lastEventAt: new Date("2026-06-24T00:00:05.000Z"),
|
||||
updatedAt,
|
||||
});
|
||||
|
||||
expect(status?.message).toContain("***REDACTED***");
|
||||
expect(status?.message.length).toBeLessThanOrEqual(MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS);
|
||||
expect(status?.currentToolName).toContain("***REDACTED***");
|
||||
expect(status?.lastAssistantSnippet).toContain("***REDACTED***");
|
||||
expect(status?.lastEventAt).toEqual(new Date("2026-06-24T00:00:05.000Z"));
|
||||
expect(getHeartbeatRunRuntimeStatus("run-1", {
|
||||
companyId: "company-1",
|
||||
issueId: "issue-1",
|
||||
|
|
@ -38,6 +45,9 @@ describe("heartbeat run runtime status store", () => {
|
|||
agentId: "agent-1",
|
||||
runId: "run-1",
|
||||
phase: "config_sync",
|
||||
currentToolName: expect.stringContaining("***REDACTED***"),
|
||||
lastAssistantSnippet: expect.stringContaining("***REDACTED***"),
|
||||
lastEventAt: new Date("2026-06-24T00:00:05.000Z"),
|
||||
});
|
||||
expect(getHeartbeatRunRuntimeStatus("run-1", { companyId: "other-company" })).toBeNull();
|
||||
expect(getHeartbeatRunRuntimeStatus("run-1", {
|
||||
|
|
@ -61,6 +71,101 @@ describe("heartbeat run runtime status store", () => {
|
|||
expect(getHeartbeatRunRuntimeStatus("run-1")).toBeNull();
|
||||
});
|
||||
|
||||
it("touch refreshes timestamps while preserving the existing status context", () => {
|
||||
setHeartbeatRunRuntimeStatus({
|
||||
companyId: "company-1",
|
||||
issueId: "issue-1",
|
||||
agentId: "agent-1",
|
||||
runId: "run-1",
|
||||
phase: "run_activity",
|
||||
message: "Using Bash",
|
||||
currentToolName: "Bash",
|
||||
lastAssistantSnippet: "Running the tests",
|
||||
updatedAt: new Date("2026-06-24T00:00:00.000Z"),
|
||||
lastEventAt: new Date("2026-06-24T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
const touched = touchHeartbeatRunRuntimeStatus({
|
||||
companyId: "company-1",
|
||||
issueId: "issue-1",
|
||||
agentId: "agent-1",
|
||||
runId: "run-1",
|
||||
at: new Date("2026-06-24T00:00:45.000Z"),
|
||||
});
|
||||
|
||||
expect(touched).toMatchObject({
|
||||
runId: "run-1",
|
||||
phase: "run_activity",
|
||||
message: "Using Bash",
|
||||
currentToolName: "Bash",
|
||||
lastAssistantSnippet: "Running the tests",
|
||||
updatedAt: new Date("2026-06-24T00:00:45.000Z"),
|
||||
lastEventAt: new Date("2026-06-24T00:00:45.000Z"),
|
||||
});
|
||||
expect(getHeartbeatRunRuntimeStatus("run-1", {
|
||||
companyId: "company-1",
|
||||
now: new Date("2026-06-24T00:02:00.000Z"),
|
||||
})).toMatchObject({ message: "Using Bash" });
|
||||
});
|
||||
|
||||
it("touch does not move timestamps backwards", () => {
|
||||
setHeartbeatRunRuntimeStatus({
|
||||
companyId: "company-1",
|
||||
issueId: null,
|
||||
agentId: "agent-1",
|
||||
runId: "run-1",
|
||||
phase: "run_activity",
|
||||
message: "Using Bash",
|
||||
updatedAt: new Date("2026-06-24T00:01:00.000Z"),
|
||||
lastEventAt: new Date("2026-06-24T00:01:00.000Z"),
|
||||
});
|
||||
|
||||
const touched = touchHeartbeatRunRuntimeStatus({
|
||||
companyId: "company-1",
|
||||
issueId: null,
|
||||
agentId: "agent-1",
|
||||
runId: "run-1",
|
||||
at: new Date("2026-06-24T00:00:30.000Z"),
|
||||
});
|
||||
|
||||
expect(touched).toMatchObject({
|
||||
updatedAt: new Date("2026-06-24T00:01:00.000Z"),
|
||||
lastEventAt: new Date("2026-06-24T00:01:00.000Z"),
|
||||
});
|
||||
});
|
||||
|
||||
it("touch creates a fallback run_activity status when none is live", () => {
|
||||
const created = touchHeartbeatRunRuntimeStatus({
|
||||
companyId: "company-1",
|
||||
issueId: "issue-1",
|
||||
agentId: "agent-1",
|
||||
runId: "run-1",
|
||||
at: new Date("2026-06-24T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
expect(created).toMatchObject({
|
||||
runId: "run-1",
|
||||
phase: "run_activity",
|
||||
message: "Receiving agent output",
|
||||
updatedAt: new Date("2026-06-24T00:00:00.000Z"),
|
||||
lastEventAt: new Date("2026-06-24T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
// An expired entry is replaced with a fresh fallback rather than revived.
|
||||
const expiredTouch = touchHeartbeatRunRuntimeStatus({
|
||||
companyId: "company-1",
|
||||
issueId: "issue-1",
|
||||
agentId: "agent-1",
|
||||
runId: "run-1",
|
||||
at: new Date("2026-06-24T00:05:00.000Z"),
|
||||
});
|
||||
expect(expiredTouch).toMatchObject({
|
||||
phase: "run_activity",
|
||||
message: "Receiving agent output",
|
||||
updatedAt: new Date("2026-06-24T00:05:00.000Z"),
|
||||
});
|
||||
});
|
||||
|
||||
it("sweeps expired statuses without touching fresh entries", () => {
|
||||
setHeartbeatRunRuntimeStatus({
|
||||
companyId: "company-1",
|
||||
|
|
|
|||
|
|
@ -3,6 +3,8 @@ 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 const MAX_HEARTBEAT_RUN_RUNTIME_TOOL_NAME_CHARS = 80;
|
||||
export const MAX_HEARTBEAT_RUN_RUNTIME_ASSISTANT_SNIPPET_CHARS = 220;
|
||||
|
||||
export interface HeartbeatRunRuntimeStatus {
|
||||
companyId: string;
|
||||
|
|
@ -12,6 +14,9 @@ export interface HeartbeatRunRuntimeStatus {
|
|||
phase: HeartbeatRunStatusPhase;
|
||||
message: string;
|
||||
updatedAt: Date;
|
||||
currentToolName: string | null;
|
||||
lastAssistantSnippet: string | null;
|
||||
lastEventAt: Date | null;
|
||||
}
|
||||
|
||||
const runtimeStatusesByRunId = new Map<string, HeartbeatRunRuntimeStatus>();
|
||||
|
|
@ -20,14 +25,27 @@ function cloneStatus(status: HeartbeatRunRuntimeStatus): HeartbeatRunRuntimeStat
|
|||
return {
|
||||
...status,
|
||||
updatedAt: new Date(status.updatedAt),
|
||||
lastEventAt: status.lastEventAt ? new Date(status.lastEventAt) : null,
|
||||
};
|
||||
}
|
||||
|
||||
export function sanitizeHeartbeatRunRuntimeStatusMessage(message: string): string {
|
||||
const normalized = message.replace(/\s+/g, " ").trim();
|
||||
function sanitizeRuntimeStatusText(value: string, maxChars: number): string {
|
||||
const normalized = value.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)}...`;
|
||||
if (redacted.length <= maxChars) return redacted;
|
||||
return `${redacted.slice(0, maxChars - 3)}...`;
|
||||
}
|
||||
|
||||
export function sanitizeHeartbeatRunRuntimeStatusMessage(message: string): string {
|
||||
return sanitizeRuntimeStatusText(message, MAX_HEARTBEAT_RUN_RUNTIME_STATUS_MESSAGE_CHARS);
|
||||
}
|
||||
|
||||
export function sanitizeHeartbeatRunRuntimeToolName(toolName: string): string {
|
||||
return sanitizeRuntimeStatusText(toolName, MAX_HEARTBEAT_RUN_RUNTIME_TOOL_NAME_CHARS);
|
||||
}
|
||||
|
||||
export function sanitizeHeartbeatRunRuntimeAssistantSnippet(snippet: string): string {
|
||||
return sanitizeRuntimeStatusText(snippet, MAX_HEARTBEAT_RUN_RUNTIME_ASSISTANT_SNIPPET_CHARS);
|
||||
}
|
||||
|
||||
function isExpired(status: HeartbeatRunRuntimeStatus, now: Date, ttlMs: number) {
|
||||
|
|
@ -35,9 +53,15 @@ function isExpired(status: HeartbeatRunRuntimeStatus, now: Date, ttlMs: number)
|
|||
}
|
||||
|
||||
export function setHeartbeatRunRuntimeStatus(
|
||||
input: Omit<HeartbeatRunRuntimeStatus, "message" | "updatedAt"> & {
|
||||
input: Omit<
|
||||
HeartbeatRunRuntimeStatus,
|
||||
"message" | "updatedAt" | "currentToolName" | "lastAssistantSnippet" | "lastEventAt"
|
||||
> & {
|
||||
message: string;
|
||||
updatedAt?: Date;
|
||||
currentToolName?: string | null;
|
||||
lastAssistantSnippet?: string | null;
|
||||
lastEventAt?: Date | null;
|
||||
},
|
||||
): HeartbeatRunRuntimeStatus | null {
|
||||
const message = sanitizeHeartbeatRunRuntimeStatusMessage(input.message);
|
||||
|
|
@ -54,11 +78,62 @@ export function setHeartbeatRunRuntimeStatus(
|
|||
phase: input.phase,
|
||||
message,
|
||||
updatedAt: input.updatedAt ? new Date(input.updatedAt) : new Date(),
|
||||
currentToolName: input.currentToolName
|
||||
? sanitizeHeartbeatRunRuntimeToolName(input.currentToolName)
|
||||
: null,
|
||||
lastAssistantSnippet: input.lastAssistantSnippet
|
||||
? sanitizeHeartbeatRunRuntimeAssistantSnippet(input.lastAssistantSnippet)
|
||||
: null,
|
||||
lastEventAt: input.lastEventAt ? new Date(input.lastEventAt) : null,
|
||||
};
|
||||
runtimeStatusesByRunId.set(status.runId, status);
|
||||
return cloneStatus(status);
|
||||
}
|
||||
|
||||
/**
|
||||
* Refresh the activity timestamps of an existing runtime status without
|
||||
* discarding its message/tool context, so streamed run-log output keeps the
|
||||
* "Working... / X ago" line fresh between structured events. When no live
|
||||
* status exists (first output, or the previous one expired past the TTL), a
|
||||
* fallback `run_activity` status is created instead.
|
||||
*/
|
||||
export function touchHeartbeatRunRuntimeStatus(input: {
|
||||
companyId: string;
|
||||
issueId: string | null;
|
||||
agentId: string;
|
||||
runId: string;
|
||||
at?: Date;
|
||||
fallbackPhase?: HeartbeatRunStatusPhase;
|
||||
fallbackMessage?: string;
|
||||
}): HeartbeatRunRuntimeStatus | null {
|
||||
const at = input.at ?? new Date();
|
||||
const existing = runtimeStatusesByRunId.get(input.runId);
|
||||
if (
|
||||
existing &&
|
||||
!isExpired(existing, at, HEARTBEAT_RUN_RUNTIME_STATUS_TTL_MS) &&
|
||||
existing.companyId === input.companyId &&
|
||||
existing.agentId === input.agentId
|
||||
) {
|
||||
if (at.getTime() > existing.updatedAt.getTime()) {
|
||||
existing.updatedAt = new Date(at);
|
||||
}
|
||||
if (!existing.lastEventAt || at.getTime() > existing.lastEventAt.getTime()) {
|
||||
existing.lastEventAt = new Date(at);
|
||||
}
|
||||
return cloneStatus(existing);
|
||||
}
|
||||
return setHeartbeatRunRuntimeStatus({
|
||||
companyId: input.companyId,
|
||||
issueId: input.issueId,
|
||||
agentId: input.agentId,
|
||||
runId: input.runId,
|
||||
phase: input.fallbackPhase ?? "run_activity",
|
||||
message: input.fallbackMessage ?? "Receiving agent output",
|
||||
updatedAt: at,
|
||||
lastEventAt: at,
|
||||
});
|
||||
}
|
||||
|
||||
export function getHeartbeatRunRuntimeStatus(
|
||||
runId: string,
|
||||
expected?: {
|
||||
|
|
|
|||
|
|
@ -204,8 +204,11 @@ import { isUnsafeSessionWorkspaceCwd } from "./session-workspace-cwd.js";
|
|||
import {
|
||||
clearHeartbeatRunRuntimeStatus,
|
||||
getHeartbeatRunRuntimeStatus,
|
||||
MAX_HEARTBEAT_RUN_RUNTIME_ASSISTANT_SNIPPET_CHARS,
|
||||
MAX_HEARTBEAT_RUN_RUNTIME_TOOL_NAME_CHARS,
|
||||
setHeartbeatRunRuntimeStatus,
|
||||
sweepExpiredHeartbeatRunRuntimeStatuses,
|
||||
touchHeartbeatRunRuntimeStatus,
|
||||
} from "./heartbeat-run-runtime-status.js";
|
||||
import {
|
||||
assertLowTrustRuntimeServicesAllowed,
|
||||
|
|
@ -266,6 +269,7 @@ export {
|
|||
ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS,
|
||||
} from "./recovery/service.js";
|
||||
export const ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS = 60 * 1000;
|
||||
export const ACTIVE_RUN_LOG_RUNTIME_STATUS_REFRESH_INTERVAL_MS = 5 * 1000;
|
||||
export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [
|
||||
2 * 60 * 1000,
|
||||
10 * 60 * 1000,
|
||||
|
|
@ -3985,6 +3989,9 @@ function decorateHeartbeatRunRuntimeStatus<T extends HeartbeatRunRuntimeStatusRu
|
|||
): T & {
|
||||
currentStatusMessage: string | null;
|
||||
currentStatusUpdatedAt: Date | null;
|
||||
currentToolName: string | null;
|
||||
lastAssistantSnippet: string | null;
|
||||
lastEventAt: Date | null;
|
||||
} {
|
||||
if (isHeartbeatRunTerminalStatus(run.status)) {
|
||||
clearHeartbeatRunRuntimeStatus(run.id);
|
||||
|
|
@ -4007,9 +4014,41 @@ function decorateHeartbeatRunRuntimeStatus<T extends HeartbeatRunRuntimeStatusRu
|
|||
...run,
|
||||
currentStatusMessage: currentStatus?.message ?? null,
|
||||
currentStatusUpdatedAt: currentStatus?.updatedAt ?? null,
|
||||
currentToolName: currentStatus?.currentToolName ?? null,
|
||||
lastAssistantSnippet: currentStatus?.lastAssistantSnippet ?? null,
|
||||
lastEventAt: currentStatus?.lastEventAt ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
function publishHeartbeatRunRuntimeProgress(status: {
|
||||
companyId: string;
|
||||
runId: string;
|
||||
agentId: string;
|
||||
issueId: string | null;
|
||||
phase: HeartbeatRunStatusPhase;
|
||||
message: string;
|
||||
updatedAt: Date;
|
||||
currentToolName?: string | null;
|
||||
lastAssistantSnippet?: string | null;
|
||||
lastEventAt?: Date | 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,
|
||||
currentToolName: status.currentToolName ?? null,
|
||||
lastAssistantSnippet: status.lastAssistantSnippet ?? null,
|
||||
lastEventAt: (status.lastEventAt ?? status.updatedAt).toISOString(),
|
||||
updatedAt: status.updatedAt.toISOString(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function recordHeartbeatRunRuntimeProgress(
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "status" | "contextSnapshot">,
|
||||
update: RuntimeStatusUpdate,
|
||||
|
|
@ -4023,24 +4062,121 @@ function recordHeartbeatRunRuntimeProgress(
|
|||
runId: run.id,
|
||||
phase: update.phase as HeartbeatRunStatusPhase,
|
||||
message: update.message,
|
||||
currentToolName: readNonEmptyString(update.currentToolName) ?? null,
|
||||
lastAssistantSnippet: readNonEmptyString(update.lastAssistantSnippet) ?? null,
|
||||
lastEventAt: update.lastEventAt ? new Date(update.lastEventAt) : new Date(),
|
||||
});
|
||||
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(),
|
||||
},
|
||||
});
|
||||
publishHeartbeatRunRuntimeProgress(status);
|
||||
return status;
|
||||
}
|
||||
|
||||
function sanitizeLiveRunProgressText(value: string, maxChars: number): string | null {
|
||||
const normalized = value.replace(/\s+/g, " ").trim();
|
||||
if (!normalized) return null;
|
||||
const redacted = redactSensitiveText(normalized);
|
||||
if (redacted.length <= maxChars) return redacted;
|
||||
return `${redacted.slice(0, maxChars - 3)}...`;
|
||||
}
|
||||
|
||||
function readLiveRunProgressString(value: unknown, maxChars: number): string | null {
|
||||
return typeof value === "string" ? sanitizeLiveRunProgressText(value, maxChars) : null;
|
||||
}
|
||||
|
||||
function readFirstLiveRunProgressString(maxChars: number, values: unknown[]): string | null {
|
||||
for (const value of values) {
|
||||
const text = readLiveRunProgressString(value, maxChars);
|
||||
if (text) return text;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function readLiveRunToolName(payload: Record<string, unknown> | null, eventType: string): string | null {
|
||||
const toolCall = parseObject(payload?.tool_call) ?? parseObject(payload?.toolCall);
|
||||
const message = parseObject(payload?.message);
|
||||
const direct = readFirstLiveRunProgressString(MAX_HEARTBEAT_RUN_RUNTIME_TOOL_NAME_CHARS, [
|
||||
payload?.toolName,
|
||||
payload?.tool_name,
|
||||
payload?.tool,
|
||||
payload?.name,
|
||||
payload?.title,
|
||||
toolCall?.name,
|
||||
toolCall?.toolName,
|
||||
message?.name,
|
||||
message?.toolName,
|
||||
]);
|
||||
if (direct) return direct;
|
||||
|
||||
const normalizedEventType = eventType.toLowerCase();
|
||||
if (!normalizedEventType.includes("tool")) return null;
|
||||
return readLiveRunProgressString(eventType.replace(/[._-]+/g, " "), MAX_HEARTBEAT_RUN_RUNTIME_TOOL_NAME_CHARS);
|
||||
}
|
||||
|
||||
function readLiveRunAssistantSnippet(
|
||||
payload: Record<string, unknown> | null,
|
||||
eventType: string,
|
||||
message: string | null,
|
||||
): string | null {
|
||||
const normalizedEventType = eventType.toLowerCase();
|
||||
const messagePayload = parseObject(payload?.message);
|
||||
const direct = readFirstLiveRunProgressString(MAX_HEARTBEAT_RUN_RUNTIME_ASSISTANT_SNIPPET_CHARS, [
|
||||
payload?.text,
|
||||
payload?.delta,
|
||||
payload?.text_delta,
|
||||
payload?.content,
|
||||
payload?.summary,
|
||||
messagePayload?.text,
|
||||
messagePayload?.content,
|
||||
]);
|
||||
if (direct) return direct;
|
||||
|
||||
if (
|
||||
normalizedEventType.includes("assistant") ||
|
||||
normalizedEventType.includes("text_delta") ||
|
||||
normalizedEventType.includes("message.delta") ||
|
||||
normalizedEventType.includes("message_delta")
|
||||
) {
|
||||
return message ? sanitizeLiveRunProgressText(message, MAX_HEARTBEAT_RUN_RUNTIME_ASSISTANT_SNIPPET_CHARS) : null;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function buildRunEventRuntimeProgress(input: {
|
||||
eventType: string;
|
||||
message: string | null;
|
||||
payload: Record<string, unknown> | null;
|
||||
at: Date;
|
||||
}) {
|
||||
const normalizedEventType = input.eventType.toLowerCase();
|
||||
if (normalizedEventType === "lifecycle" || normalizedEventType === "adapter.invoke") {
|
||||
return null;
|
||||
}
|
||||
|
||||
const currentToolName = readLiveRunToolName(input.payload, input.eventType);
|
||||
const lastAssistantSnippet = readLiveRunAssistantSnippet(input.payload, input.eventType, input.message);
|
||||
const fallbackMessage =
|
||||
readLiveRunProgressString(input.message, MAX_HEARTBEAT_RUN_RUNTIME_ASSISTANT_SNIPPET_CHARS) ??
|
||||
sanitizeLiveRunProgressText(
|
||||
input.eventType.replace(/[._-]+/g, " "),
|
||||
MAX_HEARTBEAT_RUN_RUNTIME_ASSISTANT_SNIPPET_CHARS,
|
||||
);
|
||||
const message =
|
||||
currentToolName
|
||||
? `Using ${currentToolName}`
|
||||
: lastAssistantSnippet ?? fallbackMessage;
|
||||
|
||||
if (!message) return null;
|
||||
return {
|
||||
phase: "run_activity" as const,
|
||||
message,
|
||||
currentToolName,
|
||||
lastAssistantSnippet,
|
||||
lastEventAt: input.at,
|
||||
};
|
||||
}
|
||||
|
||||
export function buildPaperclipTaskMarkdown(input: {
|
||||
issue: {
|
||||
id: string;
|
||||
|
|
@ -6552,6 +6688,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
payload?: Record<string, unknown>;
|
||||
},
|
||||
) {
|
||||
const eventAt = new Date();
|
||||
const currentUserRedactionOptions = await getCurrentUserRedactionOptions();
|
||||
const sanitizedMessage = event.message
|
||||
? redactCurrentUserText(event.message, currentUserRedactionOptions)
|
||||
|
|
@ -6563,6 +6700,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
const sanitizedPayload = secretSanitizedPayload
|
||||
? redactCurrentUserValue(secretSanitizedPayload, currentUserRedactionOptions)
|
||||
: secretSanitizedPayload;
|
||||
const issueId = readRuntimeStatusIssueIdCandidate(run) ?? null;
|
||||
const progress = buildRunEventRuntimeProgress({
|
||||
eventType: event.eventType,
|
||||
message: sanitizedMessage ?? null,
|
||||
payload: sanitizedPayload ?? null,
|
||||
at: eventAt,
|
||||
});
|
||||
|
||||
await db.insert(heartbeatRunEvents).values({
|
||||
companyId: run.companyId,
|
||||
|
|
@ -6583,15 +6727,34 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
payload: {
|
||||
runId: run.id,
|
||||
agentId: run.agentId,
|
||||
issueId,
|
||||
seq,
|
||||
eventType: event.eventType,
|
||||
stream: event.stream ?? null,
|
||||
level: event.level ?? null,
|
||||
color: event.color ?? null,
|
||||
message: sanitizedMessage ?? null,
|
||||
currentToolName: progress?.currentToolName ?? null,
|
||||
lastAssistantSnippet: progress?.lastAssistantSnippet ?? null,
|
||||
lastEventAt: (progress?.lastEventAt ?? eventAt).toISOString(),
|
||||
payload: sanitizedPayload ?? null,
|
||||
},
|
||||
});
|
||||
if (progress && isHeartbeatRunRuntimeStatusActive(run.status)) {
|
||||
const status = setHeartbeatRunRuntimeStatus({
|
||||
companyId: run.companyId,
|
||||
issueId,
|
||||
agentId: run.agentId,
|
||||
runId: run.id,
|
||||
phase: progress.phase,
|
||||
message: progress.message,
|
||||
updatedAt: eventAt,
|
||||
currentToolName: progress.currentToolName,
|
||||
lastAssistantSnippet: progress.lastAssistantSnippet,
|
||||
lastEventAt: progress.lastEventAt,
|
||||
});
|
||||
if (status) publishHeartbeatRunRuntimeProgress(status);
|
||||
}
|
||||
}
|
||||
|
||||
async function nextRunEventSeq(runId: string) {
|
||||
|
|
@ -10489,6 +10652,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
let stderrExcerpt = "";
|
||||
let outputSeq = Number(run.lastOutputSeq ?? 0);
|
||||
let lastOutputFlushAt: Date | null = run.lastOutputAt ?? null;
|
||||
let lastLogRuntimeStatusTouchMs = 0;
|
||||
const outputProgressState: {
|
||||
pending: {
|
||||
at: Date;
|
||||
|
|
@ -10629,6 +10793,28 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
};
|
||||
await flushOutputProgress();
|
||||
|
||||
// Streamed CLI output is real run activity: keep the in-memory
|
||||
// runtime status ("Working... / X ago") fresh between structured
|
||||
// events so sandbox runs with mid-run log streaming never show a
|
||||
// minutes-stale timestamp. Throttled to avoid churning the live
|
||||
// event stream on every 250ms tail chunk.
|
||||
const logActivityAt = new Date(ts);
|
||||
if (
|
||||
isHeartbeatRunRuntimeStatusActive(run.status) &&
|
||||
logActivityAt.getTime() - lastLogRuntimeStatusTouchMs >=
|
||||
ACTIVE_RUN_LOG_RUNTIME_STATUS_REFRESH_INTERVAL_MS
|
||||
) {
|
||||
lastLogRuntimeStatusTouchMs = logActivityAt.getTime();
|
||||
const touchedStatus = touchHeartbeatRunRuntimeStatus({
|
||||
companyId: run.companyId,
|
||||
issueId,
|
||||
agentId: run.agentId,
|
||||
runId: run.id,
|
||||
at: logActivityAt,
|
||||
});
|
||||
if (touchedStatus) publishHeartbeatRunRuntimeProgress(touchedStatus);
|
||||
}
|
||||
|
||||
const payloadChunk =
|
||||
sanitizedChunk.length > MAX_LIVE_LOG_CHUNK_BYTES
|
||||
? sanitizedChunk.slice(sanitizedChunk.length - MAX_LIVE_LOG_CHUNK_BYTES)
|
||||
|
|
@ -10640,6 +10826,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
|||
payload: {
|
||||
runId: run.id,
|
||||
agentId: run.agentId,
|
||||
issueId,
|
||||
ts,
|
||||
stream,
|
||||
chunk: payloadChunk,
|
||||
|
|
|
|||
|
|
@ -38,6 +38,9 @@ export interface ActiveRunForIssue {
|
|||
outputSilence?: HeartbeatRun["outputSilence"];
|
||||
currentStatusMessage?: string | null;
|
||||
currentStatusUpdatedAt?: string | Date | null;
|
||||
currentToolName?: string | null;
|
||||
lastAssistantSnippet?: string | null;
|
||||
lastEventAt?: string | Date | null;
|
||||
}
|
||||
|
||||
export interface LiveRunForIssue {
|
||||
|
|
@ -64,6 +67,9 @@ export interface LiveRunForIssue {
|
|||
outputSilence?: HeartbeatRun["outputSilence"];
|
||||
currentStatusMessage?: string | null;
|
||||
currentStatusUpdatedAt?: string | null;
|
||||
currentToolName?: string | null;
|
||||
lastAssistantSnippet?: string | null;
|
||||
lastEventAt?: string | null;
|
||||
}
|
||||
|
||||
export interface WatchdogDecisionInput {
|
||||
|
|
|
|||
|
|
@ -3464,6 +3464,8 @@ describe("IssueChatThread", () => {
|
|||
adapterType: "codex_local",
|
||||
currentStatusMessage: "Syncing git worktree to sandbox",
|
||||
currentStatusUpdatedAt: "2026-04-06T12:00:05.000Z",
|
||||
currentToolName: "bash",
|
||||
lastEventAt: new Date(Date.now() - 2000).toISOString(),
|
||||
}}
|
||||
onAdd={async () => {}}
|
||||
enableLiveTranscriptPolling={false}
|
||||
|
|
@ -3473,8 +3475,9 @@ describe("IssueChatThread", () => {
|
|||
});
|
||||
|
||||
expect(container.textContent).toContain("Working...");
|
||||
expect(container.textContent).toContain("Syncing git worktree to sandbox");
|
||||
expect(container.querySelector('[title="Syncing git worktree to sandbox"]')).not.toBeNull();
|
||||
expect(container.textContent).toContain("Using bash");
|
||||
expect(container.textContent).not.toContain("last activity");
|
||||
expect(container.textContent).toMatch(/\d+ seconds? ago/);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
|
|
|
|||
|
|
@ -301,6 +301,59 @@ function useLiveElapsed(startMs: number | null | undefined, active: boolean): st
|
|||
return formatDurationWords(Date.now() - startMs);
|
||||
}
|
||||
|
||||
function readCustomString(custom: Record<string, unknown>, key: string): string {
|
||||
return typeof custom[key] === "string" ? custom[key].trim() : "";
|
||||
}
|
||||
|
||||
function toTimestampOrNull(value: string): number | null {
|
||||
if (!value) return null;
|
||||
const timestamp = new Date(value).getTime();
|
||||
return Number.isFinite(timestamp) ? timestamp : null;
|
||||
}
|
||||
|
||||
function IssueChatLiveRunStatusLine({
|
||||
custom,
|
||||
active,
|
||||
className,
|
||||
}: {
|
||||
custom: Record<string, unknown>;
|
||||
active: boolean;
|
||||
className?: string;
|
||||
}) {
|
||||
const currentStatusMessage = readCustomString(custom, "currentStatusMessage");
|
||||
const currentToolName = readCustomString(custom, "currentToolName");
|
||||
const lastAssistantSnippet = readCustomString(custom, "lastAssistantSnippet");
|
||||
const lastEventAt = readCustomString(custom, "lastEventAt");
|
||||
const lastEventAtMs = toTimestampOrNull(lastEventAt);
|
||||
const lastActivityElapsed = useLiveElapsed(lastEventAtMs, active);
|
||||
const lastActivityAgeMs = lastEventAtMs ? Date.now() - lastEventAtMs : null;
|
||||
|
||||
if (!active) return null;
|
||||
|
||||
const primary =
|
||||
currentToolName
|
||||
? `Using ${currentToolName}`
|
||||
: lastAssistantSnippet
|
||||
? lastAssistantSnippet
|
||||
: currentStatusMessage;
|
||||
const activityText = lastActivityElapsed
|
||||
? lastActivityAgeMs !== null && lastActivityAgeMs >= 15_000
|
||||
? `no output for ${lastActivityElapsed} - still running`
|
||||
: `${lastActivityElapsed} ago`
|
||||
: "";
|
||||
const text = [primary, activityText].filter(Boolean).join(" · ");
|
||||
if (!text) return null;
|
||||
|
||||
return (
|
||||
<span
|
||||
className={cn("mt-0.5 block truncate text-xs leading-4 text-muted-foreground/70", className)}
|
||||
title={text}
|
||||
>
|
||||
{text}
|
||||
</span>
|
||||
);
|
||||
}
|
||||
|
||||
function useStableEvent<T extends (...args: never[]) => unknown>(callback: T | undefined): T | undefined {
|
||||
const callbackRef = useRef(callback);
|
||||
useLayoutEffect(() => {
|
||||
|
|
@ -873,8 +926,6 @@ function IssueChatChainOfThought({
|
|||
const rawSegments = Array.isArray(custom.chainOfThoughtSegments)
|
||||
? (custom.chainOfThoughtSegments as SegmentTiming[])
|
||||
: [];
|
||||
const currentStatusMessage =
|
||||
typeof custom.currentStatusMessage === "string" ? custom.currentStatusMessage.trim() : "";
|
||||
const segmentTiming = myIndex >= 0 ? rawSegments[myIndex] ?? null : null;
|
||||
const isActive = isCoTSegmentActive({
|
||||
isMessageRunning,
|
||||
|
|
@ -937,14 +988,7 @@ function IssueChatChainOfThought({
|
|||
<span className="text-xs text-muted-foreground/40">· {toolSummary}</span>
|
||||
) : null}
|
||||
</div>
|
||||
{isActive && currentStatusMessage ? (
|
||||
<span
|
||||
className="mt-0.5 block truncate pl-6 text-xs leading-4 text-muted-foreground/70"
|
||||
title={currentStatusMessage}
|
||||
>
|
||||
{currentStatusMessage}
|
||||
</span>
|
||||
) : null}
|
||||
<IssueChatLiveRunStatusLine custom={custom} active={isActive} className="pl-6" />
|
||||
</div>
|
||||
{hasContent ? (
|
||||
<ChevronDown className={cn("mt-0.5 h-4 w-4 shrink-0 text-muted-foreground/50 transition-transform", expanded && "rotate-180")} />
|
||||
|
|
@ -1592,8 +1636,6 @@ function IssueChatAssistantMessage({
|
|||
? custom.notices.filter((notice): notice is string => typeof notice === "string" && notice.length > 0)
|
||||
: [];
|
||||
const waitingText = typeof custom.waitingText === "string" ? custom.waitingText : "";
|
||||
const currentStatusMessage =
|
||||
typeof custom.currentStatusMessage === "string" ? custom.currentStatusMessage.trim() : "";
|
||||
const isRunning = message.role === "assistant" && message.status?.type === "running";
|
||||
const runHref = runId && runAgentId ? `/agents/${runAgentId}/runs/${runId}` : null;
|
||||
const canStopRun = Boolean(runId) && (isRunActive || runStatus === "queued" || runStatus === "running");
|
||||
|
|
@ -1881,14 +1923,7 @@ function IssueChatAssistantMessage({
|
|||
<span className="shimmer-text">{waitingText}</span>
|
||||
</span>
|
||||
</div>
|
||||
{isRunning && currentStatusMessage ? (
|
||||
<div
|
||||
className="mt-0.5 truncate pl-6 text-xs leading-4 text-muted-foreground/70"
|
||||
title={currentStatusMessage}
|
||||
>
|
||||
{currentStatusMessage}
|
||||
</div>
|
||||
) : null}
|
||||
<IssueChatLiveRunStatusLine custom={custom} active={isRunning} className="pl-6" />
|
||||
</div>
|
||||
) : null}
|
||||
{notices.length > 0 ? (
|
||||
|
|
@ -4215,6 +4250,9 @@ export function IssueChatThread({
|
|||
outputSilence: activeRun.outputSilence,
|
||||
currentStatusMessage: activeRun.currentStatusMessage ?? null,
|
||||
currentStatusUpdatedAt: toIsoString(activeRun.currentStatusUpdatedAt),
|
||||
currentToolName: activeRun.currentToolName ?? null,
|
||||
lastAssistantSnippet: activeRun.lastAssistantSnippet ?? null,
|
||||
lastEventAt: toIsoString(activeRun.lastEventAt),
|
||||
});
|
||||
}
|
||||
return [...deduped.values()].sort((a, b) => new Date(a.createdAt).getTime() - new Date(b.createdAt).getTime());
|
||||
|
|
|
|||
|
|
@ -2,11 +2,107 @@
|
|||
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { renderToStaticMarkup } from "react-dom/server";
|
||||
import { parseAcpxStdoutLine } from "@paperclipai/adapter-acpx-local/ui";
|
||||
import type { TranscriptEntry } from "../../adapters";
|
||||
import { buildTranscript, type RunLogChunk } from "../../adapters";
|
||||
import { ThemeProvider } from "../../context/ThemeContext";
|
||||
import { RunTranscriptView, normalizeTranscript } from "./RunTranscriptView";
|
||||
|
||||
describe("RunTranscriptView", () => {
|
||||
it("folds repeated tool_call status updates for the same toolUseId into one block", () => {
|
||||
const entries: TranscriptEntry[] = [
|
||||
{
|
||||
kind: "tool_call",
|
||||
ts: "2026-03-12T00:00:00.000Z",
|
||||
name: "read",
|
||||
toolUseId: "tool-1",
|
||||
input: { text: "read README.md", status: "pending" },
|
||||
},
|
||||
{
|
||||
kind: "tool_call",
|
||||
ts: "2026-03-12T00:00:01.000Z",
|
||||
name: "read",
|
||||
toolUseId: "tool-1",
|
||||
input: { status: "in_progress" },
|
||||
},
|
||||
{
|
||||
kind: "tool_call",
|
||||
ts: "2026-03-12T00:00:02.000Z",
|
||||
name: "search",
|
||||
toolUseId: "tool-2",
|
||||
input: { text: "grep TODO", status: "pending" },
|
||||
},
|
||||
{
|
||||
kind: "tool_result",
|
||||
ts: "2026-03-12T00:00:03.000Z",
|
||||
toolUseId: "tool-1",
|
||||
toolName: "read",
|
||||
content: "ok",
|
||||
isError: false,
|
||||
},
|
||||
];
|
||||
|
||||
const blocks = normalizeTranscript(entries, false);
|
||||
|
||||
expect(blocks).toHaveLength(1);
|
||||
expect(blocks[0]).toMatchObject({
|
||||
type: "tool_group",
|
||||
items: [
|
||||
{
|
||||
name: "Read",
|
||||
status: "completed",
|
||||
result: "ok",
|
||||
// Later status updates merge into the original input instead of
|
||||
// spawning duplicate "Running" cards.
|
||||
input: { text: "read README.md", status: "in_progress" },
|
||||
},
|
||||
{ name: "Search", status: "running" },
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it("renders a streamed acpx tool call as a single card once completed", () => {
|
||||
const ts = "2026-03-12T00:00:00.000Z";
|
||||
const jsonLines = [
|
||||
{ type: "acpx.tool_call", name: "read", toolCallId: "tool-1", status: "pending", text: "read README.md" },
|
||||
{ type: "acpx.tool_call", name: "read", toolCallId: "tool-1", status: "in_progress", text: "read README.md" },
|
||||
{ type: "acpx.tool_call", name: "read", toolCallId: "tool-1", status: "completed", text: "ok" },
|
||||
];
|
||||
const chunks: RunLogChunk[] = [
|
||||
{ ts, stream: "stdout", chunk: jsonLines.map((line) => JSON.stringify(line)).join("\n") + "\n" },
|
||||
];
|
||||
|
||||
const entries = buildTranscript(chunks, parseAcpxStdoutLine);
|
||||
const blocks = normalizeTranscript(entries, false);
|
||||
|
||||
expect(blocks).toHaveLength(1);
|
||||
expect(blocks[0]).toMatchObject({
|
||||
type: "tool_group",
|
||||
items: [{ name: "Read", status: "completed", result: "ok" }],
|
||||
});
|
||||
});
|
||||
|
||||
it("renders streamed acpx tool calls with input payloads as readable status cards", () => {
|
||||
const ts = "2026-03-12T00:00:00.000Z";
|
||||
const jsonLines = [
|
||||
{ type: "acpx.tool_call", name: "read", toolCallId: "tool-2", status: "running", input: { file: "README.md" } },
|
||||
{ type: "acpx.tool_call", name: "read", toolCallId: "tool-2", status: "in_progress", text: "opening file", input: { line: 1 } },
|
||||
{ type: "acpx.tool_call", name: "read", toolCallId: "tool-2", status: "completed", text: "ok" },
|
||||
];
|
||||
const chunks: RunLogChunk[] = [
|
||||
{ ts, stream: "stdout", chunk: jsonLines.map((line) => JSON.stringify(line)).join("\n") + "\n" },
|
||||
];
|
||||
|
||||
const entries = buildTranscript(chunks, parseAcpxStdoutLine);
|
||||
const blocks = normalizeTranscript(entries, false);
|
||||
|
||||
expect(blocks).toHaveLength(1);
|
||||
expect(blocks[0]).toMatchObject({
|
||||
type: "tool_group",
|
||||
items: [{ name: "Read", status: "completed", result: "ok" }],
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps running command stdout inside the command fold instead of a standalone stdout block", () => {
|
||||
const entries: TranscriptEntry[] = [
|
||||
{
|
||||
|
|
|
|||
|
|
@ -206,6 +206,21 @@ function summarizeRecord(record: Record<string, unknown>, keys: string[]): strin
|
|||
return null;
|
||||
}
|
||||
|
||||
/** Merge a streamed tool_call status update into the input captured so far. */
|
||||
function mergeToolInput(previous: unknown, incoming: unknown): unknown {
|
||||
if (incoming === null || incoming === undefined) return previous;
|
||||
if (typeof incoming === "string") {
|
||||
return incoming.trim().length > 0 ? incoming : previous;
|
||||
}
|
||||
const previousRecord = asRecord(previous);
|
||||
const incomingRecord = asRecord(incoming);
|
||||
if (incomingRecord) {
|
||||
if (Object.keys(incomingRecord).length === 0) return previous;
|
||||
return previousRecord ? { ...previousRecord, ...incomingRecord } : incoming;
|
||||
}
|
||||
return incoming;
|
||||
}
|
||||
|
||||
function summarizeToolInput(name: string, input: unknown, density: TranscriptDensity): string {
|
||||
const compactMax = density === "compact" ? 72 : 120;
|
||||
if (typeof input === "string") {
|
||||
|
|
@ -229,7 +244,7 @@ function summarizeToolInput(name: string, input: unknown, density: TranscriptDen
|
|||
|
||||
const direct =
|
||||
summarizeRecord(record, ["command", "cmd", "path", "filePath", "file_path", "query", "url", "prompt", "message"])
|
||||
?? summarizeRecord(record, ["pattern", "name", "title", "target", "tool"])
|
||||
?? summarizeRecord(record, ["pattern", "name", "title", "target", "tool", "text"])
|
||||
?? null;
|
||||
if (direct) return truncate(direct, compactMax);
|
||||
|
||||
|
|
@ -454,11 +469,20 @@ export function normalizeTranscript(entries: TranscriptEntry[], streaming: boole
|
|||
}
|
||||
|
||||
if (entry.kind === "tool_call") {
|
||||
const toolUseId = entry.toolUseId ?? extractToolUseId(entry.input);
|
||||
// Streaming runtimes (e.g. ACPX) re-emit the same tool call as its
|
||||
// status progresses. Fold updates into the existing running card
|
||||
// instead of stacking duplicate "Running" blocks.
|
||||
const pending = toolUseId ? pendingToolBlocks.get(toolUseId) : undefined;
|
||||
if (pending && pending.status === "running") {
|
||||
pending.input = mergeToolInput(pending.input, entry.input);
|
||||
continue;
|
||||
}
|
||||
const toolBlock: Extract<TranscriptBlock, { type: "tool" }> = {
|
||||
type: "tool",
|
||||
ts: entry.ts,
|
||||
name: displayToolName(entry.name, entry.input),
|
||||
toolUseId: entry.toolUseId ?? extractToolUseId(entry.input),
|
||||
toolUseId,
|
||||
input: entry.input,
|
||||
status: "running",
|
||||
};
|
||||
|
|
|
|||
|
|
@ -143,6 +143,133 @@ describe("LiveUpdatesProvider issue invalidation", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("applies heartbeat progress payloads directly to cached visible issue runs", () => {
|
||||
const cache = new Map<string, unknown>([
|
||||
[JSON.stringify(queryKeys.liveRuns("company-1")), [{ id: "run-1", currentStatusMessage: null }]],
|
||||
[JSON.stringify(queryKeys.issues.detail("DEMO-759")), {
|
||||
id: "issue-1",
|
||||
identifier: "DEMO-759",
|
||||
assigneeAgentId: "agent-1",
|
||||
}],
|
||||
[JSON.stringify(queryKeys.issues.detail("issue-1")), {
|
||||
id: "issue-1",
|
||||
identifier: "DEMO-759",
|
||||
assigneeAgentId: "agent-1",
|
||||
}],
|
||||
[JSON.stringify(queryKeys.issues.activeRun("DEMO-759")), {
|
||||
id: "run-1",
|
||||
currentStatusMessage: null,
|
||||
}],
|
||||
[JSON.stringify(queryKeys.issues.activeRun("issue-1")), {
|
||||
id: "run-1",
|
||||
currentStatusMessage: null,
|
||||
}],
|
||||
[JSON.stringify(queryKeys.issues.liveRuns("DEMO-759")), [{ id: "run-1", currentStatusMessage: null }]],
|
||||
[JSON.stringify(queryKeys.issues.liveRuns("issue-1")), [{ id: "run-1", currentStatusMessage: null }]],
|
||||
[JSON.stringify(queryKeys.issues.runs("DEMO-759")), [{ runId: "run-1" }]],
|
||||
]);
|
||||
const queryClient = {
|
||||
getQueryData: (key: unknown) => cache.get(JSON.stringify(key)),
|
||||
setQueryData: (key: unknown, updater: unknown) => {
|
||||
const cacheKey = JSON.stringify(key);
|
||||
const current = cache.get(cacheKey);
|
||||
cache.set(cacheKey, typeof updater === "function" ? updater(current) : updater);
|
||||
},
|
||||
};
|
||||
|
||||
const changed = __liveUpdatesTestUtils.applyRunLiveStatusPatchToCaches(
|
||||
queryClient as never,
|
||||
"company-1",
|
||||
"/DEMO/issues/DEMO-759",
|
||||
{
|
||||
runId: "run-1",
|
||||
agentId: "agent-1",
|
||||
issueId: "issue-1",
|
||||
message: "Syncing workspace",
|
||||
updatedAt: "2026-04-06T12:00:05.000Z",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Reading package.json",
|
||||
lastEventAt: "2026-04-06T12:00:08.000Z",
|
||||
},
|
||||
{ isForegrounded: true },
|
||||
);
|
||||
|
||||
expect(changed).toBe(true);
|
||||
expect(cache.get(JSON.stringify(queryKeys.liveRuns("company-1")))).toEqual([
|
||||
expect.objectContaining({
|
||||
id: "run-1",
|
||||
currentStatusMessage: "Syncing workspace",
|
||||
currentStatusUpdatedAt: "2026-04-06T12:00:05.000Z",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Reading package.json",
|
||||
lastEventAt: "2026-04-06T12:00:08.000Z",
|
||||
}),
|
||||
]);
|
||||
expect(cache.get(JSON.stringify(queryKeys.issues.activeRun("DEMO-759")))).toMatchObject({
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Reading package.json",
|
||||
});
|
||||
expect(cache.get(JSON.stringify(queryKeys.issues.liveRuns("issue-1")))).toEqual([
|
||||
expect.objectContaining({
|
||||
currentStatusMessage: "Syncing workspace",
|
||||
currentToolName: "bash",
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("uses the heartbeat event timestamp for run event status patches", () => {
|
||||
expect(
|
||||
__liveUpdatesTestUtils.readRunLiveStatusPatchFromPayload(
|
||||
{
|
||||
runId: "run-1",
|
||||
agentId: "agent-1",
|
||||
issueId: "issue-1",
|
||||
message: "Tool started",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Checking workspace",
|
||||
},
|
||||
"2026-04-06T12:00:09.000Z",
|
||||
"heartbeat.run.event",
|
||||
),
|
||||
).toEqual({
|
||||
runId: "run-1",
|
||||
agentId: "agent-1",
|
||||
issueId: "issue-1",
|
||||
message: "Tool started",
|
||||
updatedAt: "2026-04-06T12:00:09.000Z",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Checking workspace",
|
||||
lastEventAt: "2026-04-06T12:00:09.000Z",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not clear run tool context from null heartbeat event fields", () => {
|
||||
const patch = __liveUpdatesTestUtils.readRunLiveStatusPatchFromPayload(
|
||||
{
|
||||
runId: "run-1",
|
||||
agentId: "agent-1",
|
||||
issueId: "issue-1",
|
||||
message: null,
|
||||
currentToolName: null,
|
||||
lastAssistantSnippet: null,
|
||||
lastEventAt: "2026-04-06T12:00:10.000Z",
|
||||
},
|
||||
"2026-04-06T12:00:09.000Z",
|
||||
"heartbeat.run.event",
|
||||
);
|
||||
|
||||
expect(patch).toEqual({
|
||||
runId: "run-1",
|
||||
agentId: "agent-1",
|
||||
issueId: "issue-1",
|
||||
updatedAt: "2026-04-06T12:00:09.000Z",
|
||||
lastEventAt: "2026-04-06T12:00:10.000Z",
|
||||
});
|
||||
expect(patch).not.toHaveProperty("message");
|
||||
expect(patch).not.toHaveProperty("currentToolName");
|
||||
expect(patch).not.toHaveProperty("lastAssistantSnippet");
|
||||
});
|
||||
|
||||
it("refreshes issue document caches when a document activity event arrives", () => {
|
||||
const invalidations: unknown[] = [];
|
||||
const queryClient = {
|
||||
|
|
|
|||
|
|
@ -306,6 +306,164 @@ function invalidateVisibleIssueRunQueries(
|
|||
return true;
|
||||
}
|
||||
|
||||
interface RunLiveStatusPatch {
|
||||
runId: string;
|
||||
agentId: string | null;
|
||||
issueId: string | null;
|
||||
message?: string | null;
|
||||
updatedAt?: string | null;
|
||||
currentToolName?: string | null;
|
||||
lastAssistantSnippet?: string | null;
|
||||
lastEventAt?: string | null;
|
||||
}
|
||||
|
||||
function hasPatchKey<K extends keyof RunLiveStatusPatch>(
|
||||
patch: RunLiveStatusPatch,
|
||||
key: K,
|
||||
): patch is RunLiveStatusPatch & Required<Pick<RunLiveStatusPatch, K>> {
|
||||
return Object.prototype.hasOwnProperty.call(patch, key);
|
||||
}
|
||||
|
||||
function applyRunLiveStatusPatch<T extends ActiveRunForIssue | LiveRunForIssue>(
|
||||
run: T,
|
||||
patch: RunLiveStatusPatch,
|
||||
): T {
|
||||
if (run.id !== patch.runId) return run;
|
||||
return {
|
||||
...run,
|
||||
...(hasPatchKey(patch, "message")
|
||||
? { currentStatusMessage: patch.message }
|
||||
: {}),
|
||||
...(hasPatchKey(patch, "updatedAt")
|
||||
? { currentStatusUpdatedAt: patch.updatedAt }
|
||||
: {}),
|
||||
...(hasPatchKey(patch, "currentToolName")
|
||||
? { currentToolName: patch.currentToolName }
|
||||
: {}),
|
||||
...(hasPatchKey(patch, "lastAssistantSnippet")
|
||||
? { lastAssistantSnippet: patch.lastAssistantSnippet }
|
||||
: {}),
|
||||
...(hasPatchKey(patch, "lastEventAt")
|
||||
? { lastEventAt: patch.lastEventAt }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
function applyRunLiveStatusPatchToArray<T extends LiveRunForIssue>(
|
||||
runs: T[] | undefined,
|
||||
patch: RunLiveStatusPatch,
|
||||
): T[] | undefined {
|
||||
if (!runs) return runs;
|
||||
let changed = false;
|
||||
const nextRuns = runs.map((run) => {
|
||||
if (run.id !== patch.runId) return run;
|
||||
changed = true;
|
||||
return applyRunLiveStatusPatch(run, patch);
|
||||
});
|
||||
return changed ? nextRuns : runs;
|
||||
}
|
||||
|
||||
function readRunLiveStatusPatchFromPayload(
|
||||
payload: Record<string, unknown>,
|
||||
eventCreatedAt: string,
|
||||
eventType: string,
|
||||
): RunLiveStatusPatch | null {
|
||||
const runId = readString(payload.runId);
|
||||
if (!runId) return null;
|
||||
|
||||
const patch: RunLiveStatusPatch = {
|
||||
runId,
|
||||
agentId: readString(payload.agentId),
|
||||
issueId: readString(payload.issueId),
|
||||
};
|
||||
|
||||
if (eventType === "heartbeat.run.progress") {
|
||||
patch.message = readString(payload.message);
|
||||
patch.updatedAt = readString(payload.updatedAt) ?? eventCreatedAt;
|
||||
patch.currentToolName = readString(payload.currentToolName);
|
||||
patch.lastAssistantSnippet = readString(payload.lastAssistantSnippet);
|
||||
patch.lastEventAt = readString(payload.lastEventAt) ?? patch.updatedAt;
|
||||
return patch;
|
||||
}
|
||||
|
||||
if (eventType === "heartbeat.run.log") {
|
||||
patch.lastEventAt = readString(payload.ts) ?? eventCreatedAt;
|
||||
return patch;
|
||||
}
|
||||
|
||||
if (eventType === "heartbeat.run.event") {
|
||||
const message = readString(payload.message);
|
||||
if (message) patch.message = message;
|
||||
patch.updatedAt = readString(payload.updatedAt) ?? eventCreatedAt;
|
||||
const currentToolName = readString(payload.currentToolName);
|
||||
if (currentToolName) patch.currentToolName = currentToolName;
|
||||
const lastAssistantSnippet = readString(payload.lastAssistantSnippet);
|
||||
if (lastAssistantSnippet) patch.lastAssistantSnippet = lastAssistantSnippet;
|
||||
patch.lastEventAt = readString(payload.lastEventAt) ?? eventCreatedAt;
|
||||
return patch;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function applyRunLiveStatusPatchToCaches(
|
||||
queryClient: QueryClient,
|
||||
companyId: string,
|
||||
pathname: string,
|
||||
patch: RunLiveStatusPatch,
|
||||
options?: VisibleRouteOptions,
|
||||
): boolean {
|
||||
let changed = false;
|
||||
queryClient.setQueryData(
|
||||
queryKeys.liveRuns(companyId),
|
||||
(current: LiveRunForIssue[] | undefined) => {
|
||||
const nextRuns = applyRunLiveStatusPatchToArray(current, patch);
|
||||
if (nextRuns !== current) changed = true;
|
||||
return nextRuns;
|
||||
},
|
||||
);
|
||||
|
||||
const issueRefs = new Set<string>();
|
||||
if (patch.issueId) {
|
||||
for (const ref of resolveIssueQueryRefs(queryClient, companyId, patch.issueId, null)) {
|
||||
issueRefs.add(ref);
|
||||
}
|
||||
}
|
||||
|
||||
const context = resolveVisibleIssueRouteContext(queryClient, pathname, options);
|
||||
if (
|
||||
context &&
|
||||
(
|
||||
context.runIds.has(patch.runId) ||
|
||||
(!!patch.issueId && context.issueRefs.has(patch.issueId)) ||
|
||||
(!!patch.agentId && !!context.assigneeAgentId && patch.agentId === context.assigneeAgentId)
|
||||
)
|
||||
) {
|
||||
for (const ref of context.issueRefs) issueRefs.add(ref);
|
||||
}
|
||||
|
||||
for (const issueRef of issueRefs) {
|
||||
queryClient.setQueryData(
|
||||
queryKeys.issues.activeRun(issueRef),
|
||||
(current: ActiveRunForIssue | null | undefined) => {
|
||||
if (!current || current.id !== patch.runId) return current;
|
||||
changed = true;
|
||||
return applyRunLiveStatusPatch(current, patch);
|
||||
},
|
||||
);
|
||||
queryClient.setQueryData(
|
||||
queryKeys.issues.liveRuns(issueRef),
|
||||
(current: LiveRunForIssue[] | undefined) => {
|
||||
const nextRuns = applyRunLiveStatusPatchToArray(current, patch);
|
||||
if (nextRuns !== current) changed = true;
|
||||
return nextRuns;
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
return changed;
|
||||
}
|
||||
|
||||
function shouldSuppressAgentStatusToastForVisibleIssue(
|
||||
queryClient: QueryClient,
|
||||
pathname: string,
|
||||
|
|
@ -870,6 +1028,10 @@ function handleLiveEvent(
|
|||
|
||||
const nameOf = (id: string) => resolveAgentName(queryClient, expectedCompanyId, id);
|
||||
const payload = event.payload ?? {};
|
||||
const liveStatusPatch = readRunLiveStatusPatchFromPayload(payload, event.createdAt, event.type);
|
||||
if (liveStatusPatch) {
|
||||
applyRunLiveStatusPatchToCaches(queryClient, expectedCompanyId, pathname, liveStatusPatch);
|
||||
}
|
||||
if (event.type === "heartbeat.run.log") {
|
||||
return;
|
||||
}
|
||||
|
|
@ -979,10 +1141,12 @@ export const __liveUpdatesTestUtils = {
|
|||
buildAgentStatusToast,
|
||||
buildRunStatusToast,
|
||||
closeSocketQuietly,
|
||||
applyRunLiveStatusPatchToCaches,
|
||||
hydrateVisibleIssueComment,
|
||||
invalidateActivityQueries,
|
||||
invalidateHeartbeatProgressQueries,
|
||||
invalidateVisibleIssueRunQueries,
|
||||
readRunLiveStatusPatchFromPayload,
|
||||
resolveLiveCompanyId,
|
||||
shouldDeferIssueRefetchForVisibleAgentActivity,
|
||||
shouldDeferVisibleIssueCommentActivity,
|
||||
|
|
|
|||
|
|
@ -688,6 +688,9 @@ describe("buildIssueChatMessages", () => {
|
|||
adapterType: "codex_local",
|
||||
currentStatusMessage: "Syncing git worktree to sandbox",
|
||||
currentStatusUpdatedAt: "2026-04-06T12:03:05.000Z",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Checking repository status",
|
||||
lastEventAt: "2026-04-06T12:03:08.000Z",
|
||||
};
|
||||
|
||||
const messages = buildIssueChatMessages({
|
||||
|
|
@ -709,6 +712,9 @@ describe("buildIssueChatMessages", () => {
|
|||
runId: "run-active-1",
|
||||
currentStatusMessage: "Syncing git worktree to sandbox",
|
||||
currentStatusUpdatedAt: "2026-04-06T12:03:05.000Z",
|
||||
currentToolName: "bash",
|
||||
lastAssistantSnippet: "Checking repository status",
|
||||
lastEventAt: "2026-04-06T12:03:08.000Z",
|
||||
},
|
||||
},
|
||||
});
|
||||
|
|
|
|||
|
|
@ -871,6 +871,11 @@ function normalizeLiveRuns(
|
|||
currentStatusUpdatedAt: activeRun.currentStatusUpdatedAt
|
||||
? toDate(activeRun.currentStatusUpdatedAt).toISOString()
|
||||
: null,
|
||||
currentToolName: activeRun.currentToolName ?? null,
|
||||
lastAssistantSnippet: activeRun.lastAssistantSnippet ?? null,
|
||||
lastEventAt: activeRun.lastEventAt
|
||||
? toDate(activeRun.lastEventAt).toISOString()
|
||||
: null,
|
||||
});
|
||||
}
|
||||
return [...deduped.values()].sort((a, b) => toTimestamp(a.createdAt) - toTimestamp(b.createdAt));
|
||||
|
|
@ -911,6 +916,9 @@ function createLiveRunMessage(args: {
|
|||
chainOfThoughtSegments: segments,
|
||||
currentStatusMessage: run.currentStatusMessage ?? null,
|
||||
currentStatusUpdatedAt: run.currentStatusUpdatedAt ?? null,
|
||||
currentToolName: run.currentToolName ?? null,
|
||||
lastAssistantSnippet: run.lastAssistantSnippet ?? null,
|
||||
lastEventAt: run.lastEventAt ?? null,
|
||||
}),
|
||||
};
|
||||
return message;
|
||||
|
|
|
|||
|
|
@ -1231,6 +1231,16 @@ export function CompanyEnvironments() {
|
|||
This provider does not declare additional configuration fields.
|
||||
</div>
|
||||
)}
|
||||
<ToggleField
|
||||
label="Stream run logs"
|
||||
hint="Stream the agent CLI's output live while sandbox runs execute (recommended). Turn off to deliver output only when the run finishes."
|
||||
checked={environmentForm.sandboxConfig.streamRunLogs !== false}
|
||||
onChange={(checked) =>
|
||||
setEnvironmentForm((current) => ({
|
||||
...current,
|
||||
sandboxConfig: { ...current.sandboxConfig, streamRunLogs: checked },
|
||||
}))}
|
||||
/>
|
||||
</div>
|
||||
) : null}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue