diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 696f12b16fe..4f9f4704d6f 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -627,6 +627,8 @@ run phases: | `storage.snapshot.verify_plaintext` | Hash the decrypted archive against its sealed digest. | | `storage.snapshot.publish` | Copy the verified archive to its destination. | | `workspace.snapshot.sandbox_restore` | Upload and unpack the archive in the sandbox. | +| `workspace.snapshot.sandbox_upload` | Inside `sandbox_restore`: copy the archive into the sandbox. | +| `workspace.snapshot.sandbox_extract` | Inside `sandbox_restore`: verify and unpack the archive, then publish the restored folder. | | `storage.snapshot.prior_attempts` | Check whether an earlier upload attempt already landed. | | `storage.snapshot.encrypt` | Hash and encrypt the archive in one pass. | | `storage.snapshot.digest` | Hash the encrypted file that is uploaded. | diff --git a/server/src/__tests__/execution-preparation-phases.test.ts b/server/src/__tests__/execution-preparation-phases.test.ts index 3888f48b8a6..537e21d33e8 100644 --- a/server/src/__tests__/execution-preparation-phases.test.ts +++ b/server/src/__tests__/execution-preparation-phases.test.ts @@ -54,6 +54,20 @@ describe("run preparation phases for the run log", () => { ]); }); + it("records a step nested inside another step when an OTLP endpoint is set", async () => { + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:9"); + const runId = randomUUID(); + await createRunPreparationTiming(runId)("workspace", () => + measureRunPhase("workspace.snapshot.sandbox_restore", async () => { + await measureRunPhase("workspace.snapshot.sandbox_upload", () => delay(60)); + await measureRunPhase("workspace.snapshot.sandbox_extract", () => delay(60)); + })); + expect(drainRunPreparationPhases(runId).map(phase => phase.name)).toEqual([ + "workspace.snapshot.sandbox_upload", "workspace.snapshot.sandbox_extract", + "workspace.snapshot.sandbox_restore", "heartbeat.prepare.workspace", + ]); + }); + it("keeps runs apart", async () => { const [first, second] = [randomUUID(), randomUUID()]; await createRunPreparationTiming(first)("instructions", () => delay(60)); diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index 8c2ffa95a77..d8753197a72 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -106,6 +106,7 @@ export type ExecutionPhase = | "storage.snapshot.prior_attempts" | "storage.snapshot.encrypt" | "storage.snapshot.digest" | "storage.snapshot.put" | "storage.snapshot.verify" | "storage.snapshot.download" | "storage.snapshot.decrypt" | "storage.snapshot.verify_plaintext" | "storage.snapshot.publish" | "workspace.snapshot.sandbox_restore" + | "workspace.snapshot.sandbox_upload" | "workspace.snapshot.sandbox_extract" | "execution.runtime" | "execution.workspace_ownership" | "execution.workspace_finalize" | "execution.workspace_restore" | "execution.workspace_commit" | "execution.workspace_ownership_wait" | "execution.realtime" | "execution.usage" | "execution.native_finalize" @@ -115,7 +116,10 @@ export type ExecutionPhase = | "response.transaction" | "response.transaction.entry" | "response.transaction.commit" | "response.transaction.rollback" | "response.row_lock" | "response.write" | "response.resolve" | "response.authorize" | "response.add_comment" | "response.activity" | "response.publish"; -type RunScope = { runId: string; phase: string; parent?: Context; ended: boolean; details: Map }; +// recordRunId: the run whose preparation log receives nested measureRunPhase +// steps. A traced step opens its own scope, so it carries its preparation +// ancestor's target; other scopes (execution attempts) never record. +type RunScope = { runId: string; phase: string; parent?: Context; ended: boolean; details: Map; recordRunId?: string }; const ownership = new AsyncLocalStorage(); const endpointEnabled = () => Boolean(process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim()); @@ -186,9 +190,9 @@ export function drainRunPreparationPhases(runId: string): RecordedPreparationPha return phases; } -function measure(runId: string, name: string, work: () => T | Promise, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context): Promise { +function measure(runId: string, name: string, work: () => T | Promise, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context, recordRunId?: string): Promise { const startedAtMs = Date.now(); - const operation = measureTraced(runId, name, work, independent, counts, activeParent); + const operation = measureTraced(runId, name, work, independent, counts, activeParent, recordRunId); // Only pre-dispatch preparation reaches the run log here; execution attempts // and runtime timings belong to the turn, not to the next attempt's prepare. if (isPreparationPhaseName(name)) operation.then( @@ -198,7 +202,7 @@ function measure(runId: string, name: string, work: () => T | Promise, ind return operation; } -function measureTraced(runId: string, name: string, work: () => T | Promise, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context): Promise { +function measureTraced(runId: string, name: string, work: () => T | Promise, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context, recordRunId?: string): Promise { let operation: Promise | undefined; const run = (span?: { setStatus(status: { code: number }): void; end(): void }, parent?: Context) => { if (operation) return operation; @@ -207,6 +211,7 @@ function measureTraced(runId: string, name: string, work: () => T | Promise(), + recordRunId, }; operation = withDatabaseOperation(name, () => Promise.resolve().then(() => ownership.run(owned, () => runWithRuntimeParent(owned.parent, work))).then(value => { @@ -275,14 +280,14 @@ export function measureRunPhase(phase: ExecutionPhase, work: () => T | Promis if (count >= 16) return work(); owned.details.set(phase, count + 1); // A step inside a preparation phase reaches the run log with or without a tracer. - const recordedRunId = isPreparationPhaseName(owned.phase) ? owned.runId : null; + const recordedRunId = isPreparationPhaseName(owned.phase) ? owned.runId : owned.recordRunId ?? null; const startedAtMs = Date.now(); let outcome: RecordedPreparationPhase["outcome"] = "failed"; try { // A startup step can be more specific than the owning run phase. Keep that // causal parent while the child publishes its own context to plugin workers. const value = endpointEnabled() - ? await measure(owned.runId, phase, work, false, counts, getActiveStepContext()?.parentContext as Context | undefined) + ? await measure(owned.runId, phase, work, false, counts, getActiveStepContext()?.parentContext as Context | undefined, recordedRunId ?? undefined) : await work(); outcome = "ok"; return value; diff --git a/server/src/services/remote-workspace-transport.test.ts b/server/src/services/remote-workspace-transport.test.ts index d98970321c6..3ac0b8b7807 100644 --- a/server/src/services/remote-workspace-transport.test.ts +++ b/server/src/services/remote-workspace-transport.test.ts @@ -21,6 +21,7 @@ import { type RemoteWorkspaceTarget, } from "./remote-workspace-transport.js"; import { classifyNativeWorkspaceFailure } from "./native-runtime/native-workspace-failure.js"; +import { createRunPreparationTiming, drainRunPreparationPhases, measureRunPhase } from "./execution-preparation-timing.js"; import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; import { readWorkspaceInventory, readonlyWorkspaceInventoryScript } from "./remote-workspace-readonly-inventory.js"; @@ -241,6 +242,38 @@ describe("remote workspace transport safety with real sandbox commands", () => { } finally { await captured.cleanup(); } }); + it("times the restore's sandbox upload and extraction as separate run-log steps", async () => { + await fs.writeFile(path.join(workspace, "saved.txt"), "saved synthetic files\n"); + const { remote, execute } = target(); + const captured = await captureRemoteWorkspaceArchive({ target: remote, remoteCwd: workspace, operationId: randomUUID() }); + try { + const pause = () => new Promise(resolve => setTimeout(resolve, 60)); + const syncIn = remote.runner!.syncIn!; + remote.runner!.syncIn = async (...args) => { await pause(); return syncIn(...args); }; + const original = execute.getMockImplementation()!; + execute.mockImplementation(async input => { + if (input.args?.[0] === "-e" && input.args[1]?.includes("workspace-restored.json")) await pause(); + return original(input); + }); + // Production wraps the restore in sandbox_restore; with an OTLP endpoint + // that traced step owns the scope, and the run log must still get both. + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:9"); + const runId = randomUUID(); + await createRunPreparationTiming(runId)("workspace", () => measureRunPhase("workspace.snapshot.sandbox_restore", + () => restoreRemoteWorkspaceDirectory({ target: remote, + remoteCwd: path.join(scratch, "restored"), archivePath: captured.archivePath, + manifestDigest: captured.manifestDigest, restorationIdentity: "c".repeat(64) }))); + vi.unstubAllEnvs(); + expect(drainRunPreparationPhases(runId).map(phase => [phase.name, phase.outcome])).toEqual([ + ["workspace.snapshot.sandbox_upload", "ok"], + ["workspace.snapshot.sandbox_extract", "ok"], + ["workspace.snapshot.sandbox_restore", "ok"], + ["heartbeat.prepare.workspace", "ok"], + ]); + expect(await fs.readFile(path.join(scratch, "restored", "saved.txt"), "utf8")).toBe("saved synthetic files\n"); + } finally { await captured.cleanup(); } + }); + it("adopts an atomic restored folder after a lost reply without overwriting later edits", async () => { await fs.writeFile(path.join(workspace, "saved.txt"), "saved synthetic files\n"); const { remote, execute } = target(); diff --git a/server/src/services/remote-workspace-transport.ts b/server/src/services/remote-workspace-transport.ts index d9b09835014..6958c0a55d9 100644 --- a/server/src/services/remote-workspace-transport.ts +++ b/server/src/services/remote-workspace-transport.ts @@ -8,6 +8,7 @@ import type { RemoteWorkspaceSource } from "@paperclipai/shared"; import { captureRemoteWorkspaceDirectory, remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, remoteWorkspaceFailureDetail } from "./remote-workspace-snapshot-archive.js"; import { readWorkspaceInventory, readonlyWorkspaceInventoryScript } from "./remote-workspace-readonly-inventory.js"; import { removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js"; +import { measureRunPhase } from "./execution-preparation-timing.js"; import type { GitRemoteAuthProvider } from "./git-credentials.js"; import { classifyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-merge"; import { MAX_SANDBOX_SYNC_TIMEOUT_MS, type SandboxSyncOperation } from "@paperclipai/adapter-utils/sandbox-managed-runtime"; @@ -633,9 +634,9 @@ export async function restoreRemoteWorkspaceDirectory(input: { // Stage outside the destination; the restore validator requires an empty target. const parent = path.posix.dirname(input.remoteCwd); const staged = await installArchiveHelper(input.target, parent, operationId); - await stat(input.archivePath).then(archive => transferArchive(input.target, "syncIn", [{ operationId, + await measureRunPhase("workspace.snapshot.sandbox_upload", () => stat(input.archivePath).then(archive => transferArchive(input.target, "syncIn", [{ operationId, files: [{ kind: "file", sourcePath: input.archivePath, targetPath: staged.archive }], - }], archive.size)).catch((error: unknown) => { noteProviderRefusal("upload", error); throw error; }); + }], archive.size))).catch((error: unknown) => { noteProviderRefusal("upload", error); throw error; }); // Publish the restored folder and its identity together. A lost successful // reply can then adopt the same files without repeating a destructive restore. const restore = String.raw`const fs=require('node:fs/promises'),path=require('node:path'),cp=require('node:child_process'),crypto=require('node:crypto'); @@ -802,9 +803,9 @@ export async function restoreRemoteWorkspaceDirectory(input: { }finally{await fs.rm(staged,{recursive:true,force:true});} }finally{lock.close();} })().catch(error=>{process.stderr.write(setupFailure('remote_workspace_restore_failed',stage,error));process.exitCode=1});`; - await command(input.target, "restore", ["-e", restore, staged.helper, staged.archive, input.remoteCwd, + await measureRunPhase("workspace.snapshot.sandbox_extract", () => command(input.target, "restore", ["-e", restore, staged.helper, staged.archive, input.remoteCwd, input.manifestDigest, input.restorationIdentity, String(input.allowEmptyRoot === true), JSON.stringify(input.allocationFiles ?? []), - String(input.replaceRetiredRoot === true), input.restorationEpoch === undefined ? "" : String(input.restorationEpoch)]); + String(input.replaceRetiredRoot === true), input.restorationEpoch === undefined ? "" : String(input.restorationEpoch)])); // Publish the restored location before cleanup can fail, stall, or lose its // reply. The owner journals this receipt and then reclaims disposable staging. return { operationId, stagingRoot: parent, cleanupPending: true };