Diagnose artifact upload phases with opt-in bounded records

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