diff --git a/packages/adapter-utils/src/command-diagnostics.ts b/packages/adapter-utils/src/command-diagnostics.ts new file mode 100644 index 0000000000..afb5229c12 --- /dev/null +++ b/packages/adapter-utils/src/command-diagnostics.ts @@ -0,0 +1,18 @@ +/** Closed controller-side phases; never pass command arguments or output here. */ +export const COMMAND_DIAGNOSTIC_PHASES = [ + "chunk_queued", "global_admitted", "global_released", "upload_queued", "upload_admitted", "upload_released", + "owner_guard_start", "owner_guard_complete", "ssh_auth_prepare", "ssh_auth_ready", + "ssh_spawned", "ssh_stdin_flushed", "ssh_stdin_error", "ssh_timeout", "ssh_kill", + "ssh_exit", "ssh_close", "ssh_spawn_error", "ssh_auth_cleanup", "ssh_auth_cleaned", + "command_completed", "command_timed_out", "command_failed", "command_rejected", + "chunk_completed", "chunk_failed", +] as const; +export type CommandDiagnosticPhase = (typeof COMMAND_DIAGNOSTIC_PHASES)[number]; +export type CommandDiagnosticObserver = (phase: CommandDiagnosticPhase) => void | Promise; + +/** Diagnostics cannot grant authority, delay work, or change its outcome. */ +export function reportCommandPhase(observer: CommandDiagnosticObserver | undefined, phase: CommandDiagnosticPhase): void { + if (!observer || !COMMAND_DIAGNOSTIC_PHASES.includes(phase)) return; + try { void Promise.resolve(observer(phase)).catch(() => undefined); } + catch { /* An observer is never part of command settlement. */ } +} diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index ea013b89a2..05bc48bf92 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -1,3 +1,4 @@ +import type { CommandDiagnosticObserver } from "./command-diagnostics.js"; import { promises as fs } from "node:fs"; import { randomUUID } from "node:crypto"; import os from "node:os"; @@ -66,7 +67,7 @@ export interface CommandManagedRuntimeRunner { */ withStdinUploadSlot?( operation: () => Promise, - options: { timeoutMs: number; signal?: AbortSignal }, + options: { timeoutMs: number; signal?: AbortSignal; onPhase?: CommandDiagnosticObserver }, ): Promise; /** * True when the provider verified the concurrent-sync opt-in. A native runner @@ -93,6 +94,8 @@ export interface CommandManagedRuntimeRunner { timeoutMs?: number; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; onSpawn?: (meta: { pid: number; startedAt: string }) => Promise; + /** Opt-in closed transport stages, separate from raw command output. */ + onPhase?: CommandDiagnosticObserver; /** * Run this command through the lease's persistent session even when no run * step is active. A sandbox provider opens the session on the first diff --git a/packages/adapter-utils/src/ssh.ts b/packages/adapter-utils/src/ssh.ts index c75bf4b232..e4052b3d49 100644 --- a/packages/adapter-utils/src/ssh.ts +++ b/packages/adapter-utils/src/ssh.ts @@ -1,3 +1,4 @@ +import { reportCommandPhase, type CommandDiagnosticObserver } from "./command-diagnostics.js"; import { randomUUID } from "node:crypto"; import { execFile, spawn } from "node:child_process"; import { constants as fsConstants, createReadStream, createWriteStream, promises as fs } from "node:fs"; @@ -76,6 +77,7 @@ export function createSshCommandManagedRuntimeRunner(input: { try { const result = await runSshCommand(input.spec, remoteCommand, { stdin: commandInput.stdin, + onPhase: commandInput.onPhase, timeoutMs: commandInput.timeoutMs, maxBuffer: maxBufferBytes, }); @@ -225,6 +227,7 @@ async function spawnText( args: string[], options: { stdin?: string; + onPhase?: CommandDiagnosticObserver; timeout?: number; maxBuffer?: number; } = {}, @@ -234,6 +237,7 @@ async function spawnText( stdio: [options.stdin != null ? "pipe" : "ignore", "pipe", "pipe"], }); + reportCommandPhase(options.onPhase, "ssh_spawned"); const maxBuffer = options.maxBuffer ?? 1024 * 128; let stdout = ""; let stderr = ""; @@ -272,12 +276,14 @@ async function spawnText( const timeout = options.timeout && options.timeout > 0 ? setTimeout(() => { timedOut = true; + reportCommandPhase(options.onPhase, "ssh_timeout"); child.kill("SIGTERM"); // Escalate to SIGKILL after a 5s grace window so a hung remote // command that ignores SIGTERM cannot keep the child alive // indefinitely. killEscalation = setTimeout(() => { try { + reportCommandPhase(options.onPhase, "ssh_kill"); child.kill("SIGKILL"); } catch { // child may have already exited between the SIGTERM and the @@ -301,11 +307,14 @@ async function spawnText( }); child.on("error", (error) => { + reportCommandPhase(options.onPhase, "ssh_spawn_error"); clearTimers(); finishReject(Object.assign(error, { code: null })); }); + child.on("exit", () => reportCommandPhase(options.onPhase, "ssh_exit")); child.on("close", (code, signal) => { + reportCommandPhase(options.onPhase, "ssh_close"); clearTimers(); if (settled) return; settled = true; @@ -328,10 +337,16 @@ async function spawnText( // queued. A stream error is independent of the ChildProcess error event. // Observe it before writing, then preserve the remote exit diagnostics // when close arrives instead of letting EPIPE crash the controller. - child.stdin.on("error", (error) => { stdinError = error; }); + // This is local pipe flush, not proof of guest receipt or execution. + child.stdin.on("finish", () => reportCommandPhase(options.onPhase, "ssh_stdin_flushed")); + child.stdin.on("error", (error) => { + reportCommandPhase(options.onPhase, "ssh_stdin_error"); + stdinError = error; + }); try { child.stdin.end(options.stdin); } catch (error) { + reportCommandPhase(options.onPhase, "ssh_stdin_error"); stdinError = error instanceof Error ? error : new Error(String(error)); child.stdin.destroy(); } @@ -1293,13 +1308,16 @@ export async function runSshCommand( options: { env?: Record; stdin?: string; + onPhase?: CommandDiagnosticObserver; timeoutMs?: number; maxBuffer?: number; } = {}, ): Promise { let cleanup: () => Promise = () => Promise.resolve(); try { + reportCommandPhase(options.onPhase, "ssh_auth_prepare"); const auth = await createSshAuthArgs(config); + reportCommandPhase(options.onPhase, "ssh_auth_ready"); cleanup = auth.cleanup; const sshArgs = [...auth.args]; const envEntries = Object.entries(options.env ?? {}) @@ -1342,6 +1360,7 @@ export async function runSshCommand( return options.stdin != null ? await spawnText("ssh", sshArgs, { stdin: options.stdin, + onPhase: options.onPhase, timeout: options.timeoutMs ?? 15_000, maxBuffer: options.maxBuffer ?? 1024 * 128, }) @@ -1350,7 +1369,9 @@ export async function runSshCommand( maxBuffer: options.maxBuffer ?? 1024 * 128, }); } finally { + reportCommandPhase(options.onPhase, "ssh_auth_cleanup"); await cleanup(); + reportCommandPhase(options.onPhase, "ssh_auth_cleaned"); } } diff --git a/server/src/__tests__/ssh-artifact-diagnostics.test.ts b/server/src/__tests__/ssh-artifact-diagnostics.test.ts new file mode 100644 index 0000000000..0799e44707 --- /dev/null +++ b/server/src/__tests__/ssh-artifact-diagnostics.test.ts @@ -0,0 +1,60 @@ +import { chmod, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createSshCommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/ssh"; +import { COMMAND_DIAGNOSTIC_PHASES, reportCommandPhase, type CommandDiagnosticPhase } from "@paperclipai/adapter-utils/command-diagnostics"; + +const roots: string[] = []; +afterEach(async () => { + vi.unstubAllEnvs(); + await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true }))); +}); +async function fixture(script: string) { + const root = await mkdtemp(join(tmpdir(), "ssh-artifact-diagnostics-")); roots.push(root); + await writeFile(join(root, "ssh"), `#!${process.execPath}\n${script}\n`); + await chmod(join(root, "ssh"), 0o700); + vi.stubEnv("PATH", `${root}:${process.env.PATH}`); + return createSshCommandManagedRuntimeRunner({ spec: { + host: "private-host-canary.invalid", port: 22, username: "private-user-canary", + remoteWorkspacePath: "/private-path-canary", remoteCwd: "/private-path-canary", + privateKey: null, knownHosts: null, strictHostKeyChecking: true, + } }); +} +describe("opt-in artifact SSH phase diagnostics", () => { + it("records pipe flush, exit, close and cleanup without exposing command or output", async () => { + const runner = await fixture("process.stdin.resume(); process.stdin.on('end', () => { process.stderr.write('private-output-canary'); process.exitCode = 7; });"); + const phases: CommandDiagnosticPhase[] = []; + const result = await runner.execute({ command: "private-command-canary", args: ["private-arg-canary"], + env: { PRIVATE_TOKEN: "private-token-canary" }, stdin: "private-stdin-canary", timeoutMs: 3000, + onPhase: phase => { phases.push(phase); } }); + expect(result.exitCode).toBe(7); + expect(result.stderr).toBe("private-output-canary"); + expect(phases).toEqual(["ssh_auth_prepare", "ssh_auth_ready", "ssh_spawned", "ssh_stdin_flushed", + "ssh_exit", "ssh_close", "ssh_auth_cleanup", "ssh_auth_cleaned"]); + expect(JSON.stringify(phases)).not.toContain("canary"); + }); + it("retains the timeout and close boundary when a command never finishes", async () => { + const runner = await fixture("process.stdin.resume(); setInterval(() => {}, 1000);"); + const phases: CommandDiagnosticPhase[] = []; + const result = await runner.execute({ command: "unused", stdin: "payload", timeoutMs: 200, + onPhase: phase => { phases.push(phase); } }); + expect(result.timedOut).toBe(true); + expect(phases).toContain("ssh_timeout"); + expect(phases.indexOf("ssh_timeout")).toBeLessThan(phases.indexOf("ssh_close")); + expect(phases.slice(-2)).toEqual(["ssh_auth_cleanup", "ssh_auth_cleaned"]); + }); + it.each(["throw", "reject"])("cannot change successful execution when its observer %ss", async kind => { + const runner = await fixture("process.stdin.resume(); process.stdin.on('end', () => process.stdout.write('ok'));"); + const result = await runner.execute({ command: "unused", stdin: "payload", timeoutMs: 3000, + onPhase: () => { if (kind === "throw") throw new Error("observer canary"); return Promise.reject(new Error("observer canary")); } }); + expect(result).toMatchObject({ exitCode: 0, stdout: "ok", timedOut: false }); + }); + it("rejects unrecognized phase strings before calling the observer", () => { + const observer = vi.fn(); + reportCommandPhase(observer, "ssh_close private-path-canary" as CommandDiagnosticPhase); + expect(observer).not.toHaveBeenCalled(); + for (const phase of COMMAND_DIAGNOSTIC_PHASES) reportCommandPhase(observer, phase); + expect(observer).toHaveBeenCalledTimes(COMMAND_DIAGNOSTIC_PHASES.length); + }); +}); diff --git a/server/src/modules/computers/adapters/boat.ts b/server/src/modules/computers/adapters/boat.ts index b2bfc01e66..6e74fff226 100644 --- a/server/src/modules/computers/adapters/boat.ts +++ b/server/src/modules/computers/adapters/boat.ts @@ -1,3 +1,4 @@ +import { reportCommandPhase } from "@paperclipai/adapter-utils/command-diagnostics"; import { generateKeyPairSync, randomUUID } from "node:crypto"; import { createSshCommandManagedRuntimeRunner, @@ -85,6 +86,7 @@ export function createBoatTransportAdmission() { } const withStdinUploadSlot: NonNullable = (operation, options) => { if (uploadQueue.length >= 128) return Promise.reject(new ComputerError("conflict", "Computer upload admission is busy")); + reportCommandPhase(options.onPhase, "upload_queued"); return new Promise((resolve, reject) => { const cleanup = () => { clearTimeout(timer); options.signal?.removeEventListener("abort", aborted); }; const remove = (message: string) => { @@ -98,8 +100,10 @@ export function createBoatTransportAdmission() { const entry = { start: () => { cleanup(); uploads++; + reportCommandPhase(options.onPhase, "upload_admitted"); void Promise.resolve().then(operation).then(resolve, reject).finally(() => { uploads--; + reportCommandPhase(options.onPhase, "upload_released"); drainUploads(); }); } }; diff --git a/server/src/modules/computers/application/service.test.ts b/server/src/modules/computers/application/service.test.ts index a261e54d81..d1526691ae 100644 --- a/server/src/modules/computers/application/service.test.ts +++ b/server/src/modules/computers/application/service.test.ts @@ -116,6 +116,19 @@ function fixture() { }; } describe("computer ownership", () => { + it("forwards opt-in phases through the existing owner guard without serializing the observer", async () => { + const f = fixture(); + const phases: string[] = []; + vi.mocked(f.backend.runner).mockResolvedValue({ execute: async input => { + expect(input.stdin).not.toContain("onPhase"); + await input.onPhase?.("ssh_spawned"); + return { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", pid: null, startedAt: "" }; + } }); + await f.attach(); + const binding = await f.admit(); + await binding.process.runner.execute({ command: "fixture", onPhase: phase => { phases.push(phase); } }); + expect(phases).toEqual(["owner_guard_start", "owner_guard_complete", "ssh_spawned"]); + }); it("rechecks owner authority after a shared upload slot wait before SSH dispatch", async () => { const f = fixture(); const admission = createBoatTransportAdmission(); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 2aa863ea29..53105fc78f 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -1,3 +1,4 @@ +import { reportCommandPhase } from "@paperclipai/adapter-utils/command-diagnostics"; import { randomUUID } from "node:crypto"; import type { ComputerRepository, ComputerBackend } from "./ports.js"; import { @@ -229,12 +230,14 @@ export function createComputerService( ): CommandManagedRuntimeRunner => ({ withStdinUploadSlot: (processScoped ? control : raw).withStdinUploadSlot, execute: async (input) => { + reportCommandPhase(input.onPhase, "owner_guard_start"); const current = await (processScoped ? processScope() : scoped({ ...record, owner: ownerRef })); if ((!["active", "starting", "warm"].includes(current.owner.phase) || expired(current.owner, now())) && !(processScoped && pinnedProcess && processControlAllowed(current.owner))) throw new ComputerError("conflict", "Computer owner is retired"); + reportCommandPhase(input.onPhase, "owner_guard_complete"); // Start a command unit under the same owner slice while holding the remote tombstone lock. // The lock is released only after systemd knows the unit, so retirement cannot miss a delayed launch. return (processScoped ? control : raw).execute({ @@ -320,6 +323,7 @@ finally: timeoutMs: input.timeoutMs, onLog: input.onLog, onSpawn: input.onSpawn, + onPhase: input.onPhase, }); }, }); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index e05a130661..54c94047bf 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -1,3 +1,4 @@ +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; import { createBoatTransportAdmission } from "../../modules/computers/adapters/boat.js"; import { closeWarmNativeSession, compareAndSealRetiredComputerCheckpoint, retiredComputerCheckpointIsSettled, retiredComputerCheckpointUnsettledReason, retireUnlaunchedComputerAttempt, nativeSessionBootstrapTimeoutMs } from "./native-session-executor.js"; import { prepareHeartbeatGitHubLaunchers } from "../heartbeat-github-launchers.js"; @@ -2718,6 +2719,45 @@ describe("remote provider checkpoint restores", () => { } finally { await rm(root, { recursive: true, force: true }); } }); + it.each([{ enabled: false, rejects: false }, { enabled: true, rejects: false }, { enabled: true, rejects: true }])("records every artifact chunk outcome with private-safe diagnostics (enabled=$enabled, observer rejects=$rejects)", async ({ enabled, rejects }) => { + vi.stubEnv("PAPERCLIP_RUNNER_ARTIFACT_UPLOAD_DIAGNOSTICS", enabled ? "1" : ""); + const root = await mkdtemp(join(tmpdir(), "artifact-phase-outcomes-")); + const sourcePath = join(root, "private-source-canary"); + await writeFile(sourcePath, Buffer.alloc(9 * 1024 * 1024)); + const lines: string[] = []; + let chunk = 0; + const admission = createBoatTransportAdmission(); + const result = { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "", pid: null, startedAt: "" }; + const runner: CommandManagedRuntimeRunner = { + withStdinUploadSlot: admission.withStdinUploadSlot, + execute: async request => { + if (!request.stdin) return result; + await request.onPhase?.("private-phase-canary" as never); + const index = chunk++; + if (index === 2) throw new Error("private-exception-canary"); + return { ...result, exitCode: index === 0 ? null : 1, timedOut: index === 0, stderr: "private-stderr-canary" }; + }, + }; + try { + await expect(stageRemoteRunnerFile({ target: { kind: "remote", transport: "computer" } as never, + runner, sourcePath, targetPath: "/private-target-canary", mode: 0o700, + onLog: async (_stream, text) => { lines.push(text); if (rejects) throw new Error("private-observer-canary"); }, + })).rejects.toThrow("runner_remote_staging_failed"); + if (!enabled) { expect(lines).toEqual([]); return; } + const events = lines.map(line => JSON.parse(line.slice(line.indexOf("{")))); + expect(events.filter(event => event.phase === "chunk_failed").map(event => event.chunk)).toEqual([0, 1, 2]); + const outcomes = events.filter(event => event.phase === "chunk_failed"); + for (const phase of ["command_timed_out", "command_failed", "command_rejected"]) { + expect(outcomes.filter(event => event.phaseTimes[phase] !== undefined)).toHaveLength(1); + } + expect(outcomes.every(event => event.phaseTimes.upload_released !== undefined && event.phaseTimes.global_released !== undefined)).toBe(true); + expect(events).toHaveLength(9); // queued + admitted + final for each chunk + expect(outcomes.flatMap(event => Object.values(event.phaseTimes)).every(ms => Number.isInteger(ms) && Number(ms) >= 0 && Number(ms) <= 3_600_000)).toBe(true); + expect(events.every(event => Number.isInteger(event.elapsedMs) && event.elapsedMs >= 0 && event.elapsedMs <= 3_600_000)).toBe(true); + expect(lines.join("")).not.toContain("canary"); + } finally { vi.unstubAllEnvs(); await rm(root, { recursive: true, force: true }); } + }); + it("shares Boat upload capacity across concurrent packs before command deadlines begin", async () => { const root = await mkdtemp(join(tmpdir(), "paperclip-boat-upload-pressure-")); const sourcePath = join(root, "source"); @@ -2806,6 +2846,8 @@ describe("remote provider checkpoint restores", () => { }); it.each([{ sizeMiB: 0, corrupt: false }, { sizeMiB: 65, corrupt: false }, { sizeMiB: 0, corrupt: true }])("stages a $sizeMiB MiB provider pack through a command-only computer runner (corrupt=$corrupt)", async ({ sizeMiB, corrupt }) => { + vi.stubEnv("PAPERCLIP_RUNNER_ARTIFACT_UPLOAD_DIAGNOSTICS", "1"); + const lines: string[] = []; const root = await mkdtemp(join(tmpdir(), "paperclip-computer-pack-")); const sourcePath = join(root, "source"); const targetPath = join(root, "remote pack's files"); @@ -2843,6 +2885,7 @@ describe("remote provider checkpoint restores", () => { targetPath, mode: 0o700, onProgress, + onLog: async (_stream, text) => { lines.push(text); }, }); if (corrupt) { await expect(staging).rejects.toThrow(); @@ -2851,6 +2894,10 @@ describe("remote provider checkpoint restores", () => { return; } await staging; + const outcomes = lines.map(line => JSON.parse(line.slice(line.indexOf("{")))) + .filter(event => event.phase === "chunk_completed"); + expect(outcomes).toHaveLength(sizeMiB ? 17 : 1); + expect(outcomes.every(event => event.phaseTimes.command_completed !== undefined)).toBe(true); expect(onProgress.mock.calls[0]![0]).toBe(0); expect(onProgress.mock.calls.at(-1)![0]).toBe(onProgress.mock.calls.at(-1)![1]); expect(await readFile(join(targetPath, "provider-pack.json"), "utf8")).toBe(manifest); @@ -2867,6 +2914,7 @@ describe("remote provider checkpoint restores", () => { expect(execute.mock.calls[0]![0]).toMatchObject({ bypassSession: true }); expect((await readdir(root)).some((name) => name.includes(".upload-"))).toBe(false); } finally { + vi.unstubAllEnvs(); await rm(root, { recursive: true, force: true }); } }, 60_000); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 5d82abd6e3..0aae28988b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { COMMAND_DIAGNOSTIC_PHASES, reportCommandPhase, type CommandDiagnosticObserver } from "@paperclipai/adapter-utils/command-diagnostics"; import { computerService } from "../../modules/computers/index.js"; import { readNativeComputerWorkspaceReference } from "./native-workspace-sync.js"; import { adapterExecutionTargetIsCommandBacked, type AdapterComputerExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; @@ -10509,6 +10510,7 @@ export async function stageRemoteRunnerFile(input: { sourcePath: string; targetPath: string; mode: number; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; }): Promise { const runner = input.runner; if (runner.syncIn) { @@ -10528,11 +10530,15 @@ export async function stageRemoteRunnerFile(input: { return; } const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`; - const run = async (script: string, stdin?: string) => { + const run = async (script: string, stdin?: string, onPhase?: CommandDiagnosticObserver) => { const result = await runner.execute({ - command: "sh", args: ["-c", script], stdin, + command: "sh", args: ["-c", script], stdin, onPhase, bypassSession: true, timeoutMs: 180_000, + }).catch(error => { + reportCommandPhase(onPhase, "command_rejected"); + throw error; }); + reportCommandPhase(onPhase, result.timedOut ? "command_timed_out" : result.exitCode !== 0 ? "command_failed" : "command_completed"); if (result.exitCode !== 0 || result.timedOut) { throw new Error(`runner_remote_staging_failed: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`); } @@ -10545,7 +10551,7 @@ export async function stageRemoteRunnerFile(input: { } const remoteUpload = `${input.targetPath}.upload-${randomUUID()}`; try { - await uploadRemoteRunnerFileChunks({ sourcePath: input.sourcePath, remotePath: remoteUpload, run, runner }); + await uploadRemoteRunnerFileChunks({ sourcePath: input.sourcePath, remotePath: remoteUpload, run, runner, onLog: input.onLog }); await run(`chmod ${input.mode.toString(8)} ${quote(remoteUpload)} && mv -f ${quote(remoteUpload)} ${quote(input.targetPath)}`); } finally { await run(`rm -f ${quote(remoteUpload)}`).catch(() => undefined); @@ -10575,7 +10581,8 @@ async function uploadRemoteRunnerFileChunks(input: { runner: CommandManagedRuntimeRunner; sourcePath: string; remotePath: string; - run: (script: string, stdin?: string) => Promise; + run: (script: string, stdin?: string, onPhase?: CommandDiagnosticObserver) => Promise; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; onProgress?: (completedBytes: number, totalBytes: number) => Promise; }): Promise { const { run, remotePath } = input; @@ -10598,12 +10605,40 @@ async function uploadRemoteRunnerFileChunks(input: { // Disjoint fixed offsets permit bounded parallel upload and idempotent // writes. The final digest rejects missing, duplicated, or partial bytes. const offset = chunkIndex++; + const queuedAt = Date.now(); + // Opt-in controller diagnosis only. Keep one bounded timing per closed + // phase and a compact final record; never await a diagnostic log write. + const phaseTimes: Partial> = {}; + const onPhase: CommandDiagnosticObserver | undefined = input.onLog && + process.env.PAPERCLIP_RUNNER_ARTIFACT_UPLOAD_DIAGNOSTICS === "1" ? phase => { + if (!COMMAND_DIAGNOSTIC_PHASES.includes(phase) || phaseTimes[phase] !== undefined) return; + const elapsedMs = Math.min(3_600_000, Math.max(0, Math.trunc(Date.now() - queuedAt))); + phaseTimes[phase] = elapsedMs; + const final = phase === "chunk_completed" || phase === "chunk_failed"; + const snapshot = final || phase === "ssh_timeout" || phase === "ssh_kill"; + if (!final && phase !== "chunk_queued" && phase !== "upload_admitted" && + phase !== "ssh_spawned" && phase !== "ssh_timeout" && phase !== "ssh_kill") return; + return input.onLog!("stderr", `[paperclip-runner] artifact_upload ${JSON.stringify({ + chunk: offset, phase, elapsedMs, ...(snapshot ? { phaseTimes } : {}), + })}\n`); + } : undefined; const upload = () => run( `base64 -d | dd of=${quote(remotePath)} bs=4194304 seek=${offset} conv=notrunc 2>/dev/null`, - chunk.toString("base64")); - pending.push(withRemoteArtifactUploadSlot(() => input.runner.withStdinUploadSlot - ? input.runner.withStdinUploadSlot(upload, { timeoutMs: 180_000 }) - : upload()).catch((error) => { uploadError ??= error; })); + chunk.toString("base64"), onPhase); + reportCommandPhase(onPhase, "chunk_queued"); + pending.push(withRemoteArtifactUploadSlot(() => { + reportCommandPhase(onPhase, "global_admitted"); + return input.runner.withStdinUploadSlot + ? input.runner.withStdinUploadSlot(upload, { timeoutMs: 180_000, onPhase }) + : upload(); + }).then(() => { + reportCommandPhase(onPhase, "global_released"); + reportCommandPhase(onPhase, "chunk_completed"); + }, (error) => { + reportCommandPhase(onPhase, "global_released"); + reportCommandPhase(onPhase, "chunk_failed"); + uploadError ??= error; + })); if (pending.length === REMOTE_ARTIFACT_UPLOAD_CONCURRENCY) { await Promise.all(pending); pending.length = 0; @@ -10651,6 +10686,7 @@ export async function stageRemoteRunnerDirectory(input: { sourcePath: string; targetPath: string; mode: number; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; excludeEntries?: readonly string[]; onProgress?: (completedBytes: number, totalBytes: number) => Promise; }): Promise { @@ -10694,11 +10730,15 @@ export async function stageRemoteRunnerDirectory(input: { const archivePath = join(stagingRoot, "payload.tar.gz"); const remoteArchive = `${input.targetPath}.upload-${randomUUID()}.tar.gz`; const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`; - const run = async (script: string, stdin?: string) => { + const run = async (script: string, stdin?: string, onPhase?: CommandDiagnosticObserver) => { const result = await input.runner.execute({ - command: "sh", args: ["-c", script], stdin, + command: "sh", args: ["-c", script], stdin, onPhase, bypassSession: true, timeoutMs: 180_000, + }).catch(error => { + reportCommandPhase(onPhase, "command_rejected"); + throw error; }); + reportCommandPhase(onPhase, result.timedOut ? "command_timed_out" : result.exitCode !== 0 ? "command_failed" : "command_completed"); if (result.exitCode !== 0 || result.timedOut) { throw new Error(`runner_remote_directory_staging_failed: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`); } @@ -10712,7 +10752,7 @@ export async function stageRemoteRunnerDirectory(input: { (error) => error ? reject(error) : resolve()); }); await uploadRemoteRunnerFileChunks({ - sourcePath: archivePath, remotePath: remoteArchive, run, runner: input.runner, onProgress: input.onProgress, + sourcePath: archivePath, remotePath: remoteArchive, run, runner: input.runner, onProgress: input.onProgress, onLog: input.onLog, }); await run(`umask 077 && mkdir -p ${quote(input.targetPath)} && ` + `tar -xzf ${quote(remoteArchive)} -C ${quote(input.targetPath)} && chmod ${input.mode.toString(8)} ${quote(input.targetPath)}`); @@ -12596,6 +12636,7 @@ async function createRunnerdBackendWithinSessionClaim( target: remoteTarget, runner: remoteCommandRunner, sourcePath: sourceBinary, + onLog: input.onLog, targetPath: remoteBinary, mode: 0o700, }); @@ -12728,6 +12769,7 @@ async function createRunnerdBackendWithinSessionClaim( sourcePath: configuredProviderPackRoot, targetPath: root, mode: 0o700, + onLog: input.onLog, onProgress: async (completed, total) => { await input.onLog?.("stderr", `[paperclip-runner] Uploading provider pack: ${Math.round(completed / 1024 / 1024)} / ${Math.round(total / 1024 / 1024)} MiB.\n`); },