diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index cb7c4acb09..3b064cadb0 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -152,6 +152,7 @@ async function runExecutor( const sessionInputs: Record[] = []; const meta: Record[] = []; const logs: Array<{ stream: string; text: string }> = []; + const events: Array<{ eventType: string; payload?: Record }> = []; const execute = createAcpxEngineExecutor({ ...(options.prepareRemoteManagedHome ? { prepareRemoteManagedHome: options.prepareRemoteManagedHome } @@ -184,10 +185,13 @@ async function runExecutor( onMeta: async (payload: unknown) => { meta.push(payload as Record); }, + onEvent: async (event: { eventType: string; payload?: Record }) => { + events.push(event); + }, } as never); expect(result.exitCode).toBe(0); - return { logs, meta, runtimeOptions, configOptions, sessionInputs, result }; + return { logs, meta, events, runtimeOptions, configOptions, sessionInputs, result }; } describe("shared ACPX engine runtime behavior", () => { @@ -1790,11 +1794,18 @@ describe("ACPX engine remote sandbox staging seam (PR 1: workspace + cwd)", () = it("test_remote_buildRuntime_crosses_staging_seam", async () => { const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); - const { sessionInputs } = await runExecutor( + const { sessionInputs, events } = await runExecutor( { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, { authToken: "real-run-jwt", executionTarget }, ); + // Crossing the staging seam emits a per-step timing event for the sync. + const stageEvent = events.find( + (event) => event.eventType === "run.startup.step" && event.payload?.step === "stage.sync", + ); + expect(stageEvent).toBeTruthy(); + expect(typeof stageEvent!.payload?.durationMs).toBe("number"); + // Staging seam crossed exactly once, shipping the HOST worktree. expect(vi.mocked(prepareAdapterExecutionTargetRuntime)).toHaveBeenCalledTimes(1); const stageArgs = vi.mocked(prepareAdapterExecutionTargetRuntime).mock.calls[0]![0]; @@ -1921,7 +1932,7 @@ describe("ACPX engine remote managed-home seam (PR 2: per-adapter home seed)", ( it("test_remote_seam_receives_adapter_agnostic_context", async () => { const { stateDir, localCwd, remoteCwd, executionTarget } = await setupRemoteSandbox(); let captured: Record | null = null; - const { sessionInputs } = await runExecutor( + const { sessionInputs, events } = await runExecutor( { agent: "custom", agentCommand: "node ./fake-acp.js", @@ -1941,6 +1952,14 @@ describe("ACPX engine remote managed-home seam (PR 2: per-adapter home seed)", ( }, ); + // The managed-home seam runs inside the timed stage.sync boundary, so a + // per-step timing event is emitted for it. + const stageEvent = events.find( + (event) => event.eventType === "run.startup.step" && event.payload?.step === "stage.sync", + ); + expect(stageEvent).toBeTruthy(); + expect(typeof stageEvent!.payload?.durationMs).toBe("number"); + // The engine invoked the seam and used the runtime it staged (session/new // binds to the in-sandbox workspace dir the seam returned). expect(captured).not.toBeNull(); @@ -2659,3 +2678,120 @@ describe("ACPX engine remote session-lifecycle re-staging (PR 3: stage once / re expect(events).toContain("exit:run-b"); }); }); + +describe("ACPX engine per-step startup timing (run.startup.step events)", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + function stepEvents(events: Array<{ eventType: string; payload?: Record }>) { + return events.filter((event) => event.eventType === "run.startup.step"); + } + + it("emits a run.startup.step event for each of the 7 bring-up boundaries with numeric durationMs", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + // A configured CODEX_HOME keeps the codex-home seed deterministic (skips the + // managed-home copy from the host ~/.codex) so steps 2 and 3 run cleanly. + const codexHome = path.join(root, "codex-home"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + await fs.mkdir(codexHome, { recursive: true }); + const executionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + }; + + const { events } = await runExecutor( + { + agent: "codex", + agentCommand: "node ./fake-acp.js", + stateDir, + cwd: localCwd, + env: { CODEX_HOME: codexHome }, + }, + { authToken: "real-run-jwt", executionTarget }, + ); + + const steps = stepEvents(events); + const seen = new Map(steps.map((event) => [String(event.payload?.step), event])); + // A codex bring-up over the remote sandbox lane crosses all 7 boundaries. + for (const step of [ + "workspace.resolve", + "codex-home.seed", + "skills.reconcile", + "stage.sync", + "bridge.paperclip", + "bridge.process-session", + "acp.handshake", + ]) { + const event = seen.get(step); + expect(event, `expected a run.startup.step event for "${step}"`).toBeTruthy(); + expect(typeof event!.payload?.durationMs).toBe("number"); + expect(event!.payload?.durationMs as number).toBeGreaterThanOrEqual(0); + } + }); + + it("emits the 5 non-codex boundaries for a custom-agent sandbox bring-up (no codex steps)", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const localCwd = path.join(root, "worktree"); + const remoteCwd = path.join(root, "remote-workspace"); + await fs.mkdir(localCwd, { recursive: true }); + await fs.mkdir(remoteCwd, { recursive: true }); + const executionTarget = { + kind: "remote", + transport: "sandbox", + providerKey: "fake-plugin", + remoteCwd, + runner: createLocalSandboxRunner(), + }; + + const { events } = await runExecutor( + { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir, cwd: localCwd }, + { authToken: "real-run-jwt", executionTarget }, + ); + + const emitted = new Set(stepEvents(events).map((event) => String(event.payload?.step))); + // The custom-agent lane skips the codex-only skill prep entirely... + expect(emitted.has("codex-home.seed")).toBe(false); + expect(emitted.has("skills.reconcile")).toBe(false); + // ...but still times the shared workspace/stage/bridge/handshake boundaries. + for (const step of [ + "workspace.resolve", + "stage.sync", + "bridge.paperclip", + "bridge.process-session", + "acp.handshake", + ]) { + expect(emitted.has(step), `expected a run.startup.step event for "${step}"`).toBe(true); + } + }); + + it("does not emit startup-step events on a local (non-sandbox) run except workspace.resolve", async () => { + const root = await makeTempRoot(); + const localCwd = path.join(root, "worktree"); + await fs.mkdir(localCwd, { recursive: true }); + + const { events } = await runExecutor({ + agent: "custom", + agentCommand: "node ./fake-acp.js", + stateDir: path.join(root, "state"), + cwd: localCwd, + }); + + const emitted = new Set(stepEvents(events).map((event) => String(event.payload?.step))); + // A local run never crosses the staging seam or starts a bridge, so only the + // always-run workspace resolution and the ACP handshake are timed. + expect(emitted.has("workspace.resolve")).toBe(true); + expect(emitted.has("acp.handshake")).toBe(true); + expect(emitted.has("stage.sync")).toBe(false); + expect(emitted.has("bridge.paperclip")).toBe(false); + expect(emitted.has("bridge.process-session")).toBe(false); + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index fddf84b0ac..9c89e70df6 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -81,6 +81,7 @@ import { DEFAULT_ACP_ENGINE_TIMEOUT_SEC, DEFAULT_ACP_ENGINE_WARM_HANDLE_IDLE_MS, } from "./constants.js"; +import { measureStartupStep } from "./startup-timing.js"; const defaultModuleDir = path.dirname(fileURLToPath(import.meta.url)); const PAPERCLIP_MANAGED_CODEX_SKILLS_MANIFEST = ".paperclip-managed-skills.json"; @@ -829,7 +830,15 @@ async function prepareCodexSkillRuntime(input: { env: Record; moduleDir: string; onLog: AdapterExecutionContext["onLog"]; + // Step-timing seam: threaded from `buildRuntime` so the nested + // `skills.reconcile` boundary (step 3) can emit its own `run.startup.step` + // event at its call-site. Both optional — a caller without an event sink or + // clock is a plain no-op passthrough (the timing helper guards a missing + // `onEvent`), so the codex skill prep behaves identically when unmeasured. + onEvent?: AdapterExecutionContext["onEvent"]; + now?: () => number; }): Promise<{ identity: Record; commandNotes: string[] }> { + const now = input.now ?? (() => Date.now()); const envConfig = parseObject(input.config.env); const configuredCodexHome = typeof envConfig.CODEX_HOME === "string" && envConfig.CODEX_HOME.trim().length > 0 @@ -851,12 +860,16 @@ async function prepareCodexSkillRuntime(input: { const skillSetKey = await buildSkillSetKey({ skills: selectedSkills, label: "codex" }); const skillsHome = path.join(effectiveCodexHome, "skills"); await fs.mkdir(skillsHome, { recursive: true }); - await reconcileManagedCodexSkills({ - skillsHome, - allSkills, - selectedSkills, - onLog: input.onLog, - }); + // Step 3 — skills.reconcile: nested inside the codex-home seed (step 2), so it + // emits its own boundary event at this call-site. + await measureStartupStep({ onEvent: input.onEvent }, now, "skills.reconcile", () => + reconcileManagedCodexSkills({ + skillsHome, + allSkills, + selectedSkills, + onLog: input.onLog, + }), + ); for (const entry of selectedSkills) { const target = path.join(skillsHome, entry.runtimeName); @@ -1208,6 +1221,10 @@ async function buildRuntime(input: { deps: AcpxEngineExecutorOptions; }): Promise { const { runId, agent, config, context, authToken } = input.ctx; + // Injectable monotonic clock for per-step startup timing. Hoisted above the + // first instrumented boundary (step 1 `workspace.resolve`, below) so every + // `measureStartupStep` call in this function shares one deterministic clock. + const nowMs = input.deps.now ?? (() => Date.now()); const workspaceContext = parseObject(context.paperclipWorkspace); const secretsContext = parseObject(context.paperclipSecrets); const secretManifest = Array.isArray(secretsContext.manifest) ? secretsContext.manifest : []; @@ -1240,7 +1257,11 @@ async function buildRuntime(input: { executionTargetIsRemote, executionCwd: effectiveExecutionCwd, }); - await ensureAbsoluteDirectory(cwd, { createIfMissing: true }); + // Step 1 — workspace.resolve: the workspace resolution/fallback chain closes + // here on the awaited directory materialization. + await measureStartupStep(input.ctx, nowMs, "workspace.resolve", () => + ensureAbsoluteDirectory(cwd, { createIfMissing: true }), + ); const acpxAgent = normalizeAgent(config); const mode = normalizeMode(config); @@ -1388,13 +1409,20 @@ async function buildRuntime(input: { }, +${paperclipClaudeSettings.additionalDirectories.length} read root(s), +${paperclipClaudeSettings.allow.length} allow rule(s)).`, ); } else if (acpxAgent === "codex") { - const preparedSkills = await prepareCodexSkillRuntime({ - companyId: agent.companyId, - config, - env, - moduleDir: input.engine.moduleDir, - onLog: input.ctx.onLog, - }); + // Step 2 — codex-home.seed: the codex managed-home + skills preparation. + // The nested skills.reconcile boundary (step 3) is timed inside via the + // threaded onEvent/now seam. + const preparedSkills = await measureStartupStep(input.ctx, nowMs, "codex-home.seed", () => + prepareCodexSkillRuntime({ + companyId: agent.companyId, + config, + env, + moduleDir: input.engine.moduleDir, + onLog: input.ctx.onLog, + onEvent: input.ctx.onEvent, + now: nowMs, + }), + ); skillsIdentity = preparedSkills.identity; skillCommandNotes.push(...preparedSkills.commandNotes); } else if (acpxAgent === "gemini") { @@ -1534,7 +1562,6 @@ async function buildRuntime(input: { // re-checks the cache before deciding). const stagedRuntimes = input.deps.stagedRuntimes ?? defaultStagedRuntimes; const stagingLocks = input.deps.stagingLocks ?? defaultStagingLocks; - const nowMs = input.deps.now ?? (() => Date.now()); const previousParams = parseObject(input.ctx.runtime.sessionParams); const isCompatibleResume = isCompatibleSession(previousParams, { fingerprint, @@ -1612,29 +1639,42 @@ async function buildRuntime(input: { // verbatim on a later compatible resume. Add/change only — every seam sets // (never deletes) its home env var, so a set-based delta is complete. const envBeforeStage = { ...env }; - let freshStagedRuntime: PreparedAdapterExecutionTargetRuntime; - let freshTeardown: (() => Promise) | null = null; - let freshDispose: (() => Promise) | null = null; - if (input.deps.prepareRemoteManagedHome) { - const seeded = await input.deps.prepareRemoteManagedHome({ - acpxAgent, - companyId: agent.companyId, - runId, - config, - executionTarget: remoteTarget, - workspaceLocalDir: cwd, - timeoutSec, - env, - onLog: input.ctx.onLog, - onRuntimeProgress: input.ctx.onRuntimeProgress, - stage, - }); - freshStagedRuntime = seeded.stagedRuntime; - freshTeardown = seeded.teardown ?? null; - freshDispose = seeded.disposeStaged ?? null; - } else { - freshStagedRuntime = await stage([]); - } + // Step 4 — stage.sync: ship the workspace (and, via the seam, the managed + // home) into the sandbox. Only fires on a fresh stage; a compatible resume + // that reuses an already-staged runtime skips this block entirely. The + // measured callback returns the staged result so the timing wrap does not + // disturb definite-assignment of the outer bindings. + const { + stagedRuntime: freshStagedRuntime, + teardown: freshTeardown, + dispose: freshDispose, + } = await measureStartupStep(input.ctx, nowMs, "stage.sync", async (): Promise<{ + stagedRuntime: PreparedAdapterExecutionTargetRuntime; + teardown: (() => Promise) | null; + dispose: (() => Promise) | null; + }> => { + if (input.deps.prepareRemoteManagedHome) { + const seeded = await input.deps.prepareRemoteManagedHome({ + acpxAgent, + companyId: agent.companyId, + runId, + config, + executionTarget: remoteTarget, + workspaceLocalDir: cwd, + timeoutSec, + env, + onLog: input.ctx.onLog, + onRuntimeProgress: input.ctx.onRuntimeProgress, + stage, + }); + return { + stagedRuntime: seeded.stagedRuntime, + teardown: seeded.teardown ?? null, + dispose: seeded.disposeStaged ?? null, + }; + } + return { stagedRuntime: await stage([]), teardown: null, dispose: null }; + }); const delta: Record = {}; for (const [key, value] of Object.entries(env)) { if (envBeforeStage[key] !== value) delta[key] = value; @@ -1663,15 +1703,18 @@ async function buildRuntime(input: { let runtimeEnv: Record = {}; try { if (useRemoteProcessSession) { - paperclipBridge = await startAdapterExecutionTargetPaperclipBridge({ - runId, - target: { ...executionTarget, streamRunLogs: false }, - runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, - adapterKey: input.engine.adapterType, - timeoutSec, - hostApiToken: env.PAPERCLIP_API_KEY, - onLog: input.ctx.onLog, - }); + // Step 5 — bridge.paperclip: start the sandbox ACP API callback bridge. + paperclipBridge = await measureStartupStep(input.ctx, nowMs, "bridge.paperclip", () => + startAdapterExecutionTargetPaperclipBridge({ + runId, + target: { ...executionTarget, streamRunLogs: false }, + runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, + adapterKey: input.engine.adapterType, + timeoutSec, + hostApiToken: env.PAPERCLIP_API_KEY, + onLog: input.ctx.onLog, + }), + ); if (paperclipBridge) { Object.assign(env, paperclipBridge.env); await input.ctx.onLog("stdout", "[paperclip] Sandbox ACP API callback bridge enabled for this run.\n"); @@ -1682,19 +1725,22 @@ async function buildRuntime(input: { (entry): entry is [string, string] => typeof entry[1] === "string", ), ); + // Step 6 — bridge.process-session: start the in-sandbox process session. processSessionBridge = useRemoteProcessSession - ? await startAdapterExecutionTargetProcessSessionBridge({ - runId, - target: executionTarget, - runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, - adapterKey: input.engine.adapterType, - command: "sh", - args: ["-lc", `exec ${agentCommandShell}`], - cwd: sessionCwd, - env: runtimeEnv, - timeoutSec, - onLog: input.ctx.onLog, - }) + ? await measureStartupStep(input.ctx, nowMs, "bridge.process-session", () => + startAdapterExecutionTargetProcessSessionBridge({ + runId, + target: executionTarget, + runtimeRootDir: stagedRuntime?.runtimeRootDir ?? null, + adapterKey: input.engine.adapterType, + command: "sh", + args: ["-lc", `exec ${agentCommandShell}`], + cwd: sessionCwd, + env: runtimeEnv, + timeoutSec, + onLog: input.ctx.onLog, + }), + ) : null; } catch (err) { await paperclipBridge?.stop().catch(() => {}); @@ -2558,14 +2604,19 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { try { if (!handle) { try { - handle = await runtime.ensureSession({ - sessionKey: prepared.sessionKey, - agent: prepared.acpxAgent, - mode: prepared.mode, - cwd: prepared.cwd, - resumeSessionId, - sessionOptions: { env: prepared.env }, - }); + // Step 7 — acp.handshake: ACP session establishment (session/new or + // resume). A throwing handshake still reports its duration before the + // resume-retry path below runs. + handle = await measureStartupStep(ctx, now, "acp.handshake", () => + runtime.ensureSession({ + sessionKey: prepared.sessionKey, + agent: prepared.acpxAgent, + mode: prepared.mode, + cwd: prepared.cwd, + resumeSessionId, + sessionOptions: { env: prepared.env }, + }), + ); } catch (err) { if (!resumeSessionId || !isResumeFailure(err)) throw err; clearSession = true; @@ -2574,13 +2625,15 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { "stdout", `[paperclip] ACPX resume session "${resumeSessionId}" is unavailable; retrying with a fresh session.\n`, ); - handle = await runtime.ensureSession({ - sessionKey: prepared.sessionKey, - agent: prepared.acpxAgent, - mode: prepared.mode, - cwd: prepared.cwd, - sessionOptions: { env: prepared.env }, - }); + handle = await measureStartupStep(ctx, now, "acp.handshake", () => + runtime.ensureSession({ + sessionKey: prepared.sessionKey, + agent: prepared.acpxAgent, + mode: prepared.mode, + cwd: prepared.cwd, + sessionOptions: { env: prepared.env }, + }), + ); } } } catch (err) { diff --git a/packages/adapter-utils/src/acpx-engine/startup-timing.test.ts b/packages/adapter-utils/src/acpx-engine/startup-timing.test.ts new file mode 100644 index 0000000000..6db874f730 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/startup-timing.test.ts @@ -0,0 +1,116 @@ +import { describe, expect, it, vi } from "vitest"; +import type { AdapterRuntimeEvent } from "../types.js"; +import { measureStartupStep } from "./startup-timing.js"; + +describe("measureStartupStep", () => { + it("emits one run.startup.step event with the step name and measured durationMs", async () => { + let t = 0; + const now = () => t; + const events: AdapterRuntimeEvent[] = []; + const onEvent = vi.fn(async (event: AdapterRuntimeEvent) => { + events.push(event); + }); + + const result = await measureStartupStep({ onEvent }, now, "stage.sync", async () => { + t = 150; // clock advances while the wrapped step runs + return "ok"; + }); + + expect(result).toBe("ok"); + expect(onEvent).toHaveBeenCalledTimes(1); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ + eventType: "run.startup.step", + stream: "system", + level: "info", + payload: { step: "stage.sync", durationMs: 150 }, + }); + expect(events[0]!.message).toBe("startup step: stage.sync (150ms)"); + }); + + it("returns the wrapped fn result unchanged", async () => { + const now = () => 0; + const onEvent = vi.fn(async () => {}); + const value = { nested: [1, 2, 3] }; + + const result = await measureStartupStep({ onEvent }, now, "workspace.resolve", async () => value); + + expect(result).toBe(value); + }); + + it("still emits the timing event and re-throws when fn rejects", async () => { + let t = 0; + const now = () => t; + const events: AdapterRuntimeEvent[] = []; + const onEvent = vi.fn(async (event: AdapterRuntimeEvent) => { + events.push(event); + }); + const boom = new Error("step failed"); + + await expect( + measureStartupStep({ onEvent }, now, "acp.handshake", async () => { + t = 42; + throw boom; + }), + ).rejects.toBe(boom); + + expect(onEvent).toHaveBeenCalledTimes(1); + expect(events[0]).toMatchObject({ + eventType: "run.startup.step", + payload: { step: "acp.handshake", durationMs: 42 }, + }); + }); + + it("swallows onEvent errors without changing the wrapped fn result", async () => { + let t = 0; + const now = () => t; + const onEvent = vi.fn(async () => { + throw new Error("sink failed"); + }); + + const result = await measureStartupStep({ onEvent }, now, "bridge.paperclip", async () => { + t = 17; + return "value"; + }); + + expect(result).toBe("value"); + expect(onEvent).toHaveBeenCalledTimes(1); + }); + + it("swallows onEvent errors without replacing a wrapped fn error", async () => { + let t = 0; + const now = () => t; + const onEvent = vi.fn(async () => { + throw new Error("sink failed"); + }); + const boom = new Error("step failed"); + + await expect( + measureStartupStep({ onEvent }, now, "bridge.process-session", async () => { + t = 17; + throw boom; + }), + ).rejects.toBe(boom); + + expect(onEvent).toHaveBeenCalledTimes(1); + }); + + it("does not throw when ctx.onEvent is undefined", async () => { + const now = () => 0; + + await expect( + measureStartupStep({}, now, "bridge.paperclip", async () => "value"), + ).resolves.toBe("value"); + }); + + it("still surfaces the fn error when ctx.onEvent is undefined", async () => { + const now = () => 0; + const boom = new Error("undefined-sink failure"); + + await expect( + measureStartupStep({}, now, "bridge.process-session", async () => { + throw boom; + }), + ).rejects.toBe(boom); + }); +}); diff --git a/packages/adapter-utils/src/acpx-engine/startup-timing.ts b/packages/adapter-utils/src/acpx-engine/startup-timing.ts new file mode 100644 index 0000000000..06e46f1313 --- /dev/null +++ b/packages/adapter-utils/src/acpx-engine/startup-timing.ts @@ -0,0 +1,47 @@ +import type { AdapterExecutionContext, AdapterRuntimeEvent } from "../types.js"; + +/** + * Structured event emitted once per named sandbox run-startup boundary so the + * duration of each bring-up step lands in the `heartbeat_run_events` stream + * (jsonb `payload`) beside the existing "run started" / "adapter invocation" + * anchors. Observability-only — it rides the existing + * `ctx.onEvent → onAdapterEvent → appendRunEvent` bridge with no schema change. + */ +export const RUN_STARTUP_STEP_EVENT_TYPE = "run.startup.step"; + +function buildStepEvent(step: string, durationMs: number): AdapterRuntimeEvent { + return { + eventType: RUN_STARTUP_STEP_EVENT_TYPE, + stream: "system", + level: "info", + message: `startup step: ${step} (${durationMs}ms)`, + payload: { step, durationMs }, + }; +} + +/** + * Time `fn` with the injected `now` clock and emit exactly one + * `run.startup.step` event carrying `{ step, durationMs }`. The event fires in a + * `finally`, so a throwing step still reports its duration before the error is + * re-thrown. `now` is injected (never `Date.now()` here) so callers/tests stay + * deterministic, and `ctx.onEvent` is optional — a missing sink is a no-op that + * neither throws nor swallows `fn`'s return value or error. + */ +export async function measureStartupStep( + ctx: Pick, + now: () => number, + step: string, + fn: () => Promise, +): Promise { + const start = now(); + try { + return await fn(); + } finally { + const durationMs = now() - start; + try { + await ctx.onEvent?.(buildStepEvent(step, durationMs)); + } catch { + // Observability must not change startup control flow. + } + } +}