Merge branch 'codex/boat-environment-core' into codex/boat-environment

* codex/boat-environment-core:
  Diagnose artifact upload phases with opt-in bounded records

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<void>;
+
+/** 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. */ }
+}

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?<T>(
     operation: () => Promise<T>,
-    options: { timeoutMs: number; signal?: AbortSignal },
+    options: { timeoutMs: number; signal?: AbortSignal; onPhase?: CommandDiagnosticObserver },
   ): Promise<T>;
   /**
    * 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<void>;
     onSpawn?: (meta: { pid: number; startedAt: string }) => Promise<void>;
+    /** 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

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<string, string>;
     stdin?: string;
+    onPhase?: CommandDiagnosticObserver;
     timeoutMs?: number;
     maxBuffer?: number;
   } = {},
 ): Promise<SshCommandResult> {
   let cleanup: () => Promise<void> = () => 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");
   }
 }
 

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);
+  });
+});

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<CommandManagedRuntimeRunner["withStdinUploadSlot"]> = (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();
         });
       } };

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();

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,
         });
       },
     });

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);

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<void>;
 }): Promise<void> {
   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<void>;
+  run: (script: string, stdin?: string, onPhase?: CommandDiagnosticObserver) => Promise<void>;
+  onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
   onProgress?: (completedBytes: number, totalBytes: number) => Promise<void>;
 }): Promise<void> {
   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<Record<(typeof COMMAND_DIAGNOSTIC_PHASES)[number], number>> = {};
+      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<void>;
   excludeEntries?: readonly string[];
   onProgress?: (completedBytes: number, totalBytes: number) => Promise<void>;
 }): Promise<void> {
@@ -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`);
               },