feat(acpx): per-step timing observability for sandbox run-startup (#10204)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Sandbox-backed runs need observable startup behavior so operators can see where time is spent before an adapter is invoked > - The current startup path only surfaced aggregate timing, which makes it hard to identify the slow boundary in the bring-up sequence > - That gap matters because sandbox startup latency is often dominated by one specific step, and aggregate timing hides the bottleneck > - This pull request adds per-step startup timing events for the named sandbox bring-up boundaries > - The benefit is more precise observability with no control-flow change and no schema migration ## Linked Issues or Issue Description ### Subsystem affected Cross-cutting (multiple of the above) ### Problem or motivation Sandbox run startup only exposed aggregate timing. That makes it hard to identify which bring-up boundary is responsible for slow starts, especially in remote or sandboxed execution where the bottleneck can move between workspace setup, skill reconciliation, bridge setup, and adapter handshake. ### Proposed solution Emit a structured timing event for each named startup boundary before the adapter is invoked, so the existing run-event stream carries per-step duration data. This keeps the event path additive and lets operators see which step dominates startup latency without changing control flow or introducing a schema migration. ### Alternatives considered - Keep only the aggregate startup duration: simpler, but it hides the bottleneck and makes regression analysis much harder. - Add a new telemetry sink or schema field: rejected because the existing run-event payload already carries structured event data and does not need a new storage path. - Log unstructured text for each step: rejected because it is harder to query and aggregate than a structured `step` + `durationMs` event. ### Roadmap alignment This fits the roadmap direction around cloud / sandbox agents and enforced outcomes by improving observability for sandboxed execution without changing the control plane model. The roadmap section is broad, but it does not call out this specific startup-timing work as a planned duplicate. ### Additional context This PR is intentionally additive. It records timing for the named startup boundaries in the existing event stream and leaves the bridge, database shape, and adapter invocation order unchanged. ## What Changed - Added a `measureStartupStep` helper that times a startup step, emits one structured `run.startup.step` event, and rethrows failures after recording duration - Wrapped the seven sandbox bring-up boundaries in `execute.ts` so the structured timing covers each named step before adapter invocation - Added unit coverage for the helper and integration coverage for the startup-step events in the adapter-utils execute path - Kept the event path additive, with no bridge change and no database migration ## Verification - `tsc --noEmit` for `@paperclip/adapter-utils` - `pnpm test` in `packages/adapter-utils` equivalent suite coverage: 292 passed, 4 skipped - Adjacent server event/log-store suites: `run-log-store.test.ts` and `heartbeat-run-log.test.ts` passed (11 total) - Git validation: fetched `origin/feat/sandbox-startup-step-timing`, confirmed it matches the authorized submit SHA, and confirmed `origin/master..origin/feat/sandbox-startup-step-timing` contains the expected single commit - Searched GitHub for duplicate or related open PRs/issues and found no overlapping open items - Checked `ROADMAP.md`; the roadmap covers sandboxed environments generally, but does not call out this specific startup-timing observability work as a planned duplicate ## Risks - Low risk: the change is additive and only emits additional structured events - If downstream consumers assume startup events are aggregate-only, they may need to ignore or account for the new `run.startup.step` entries - Timing is measured via the injected clock and event emission happens in a `finally`, so failures still report duration before rethrowing ## Model Used OpenAI GPT-5, tool-using coding agent ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Harold Kim <harold@paperclip.ing> Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
7014e46e5b
commit
762ce5b4ef
|
|
@ -152,6 +152,7 @@ async function runExecutor(
|
|||
const sessionInputs: Record<string, unknown>[] = [];
|
||||
const meta: Record<string, unknown>[] = [];
|
||||
const logs: Array<{ stream: string; text: string }> = [];
|
||||
const events: Array<{ eventType: string; payload?: Record<string, unknown> }> = [];
|
||||
const execute = createAcpxEngineExecutor({
|
||||
...(options.prepareRemoteManagedHome
|
||||
? { prepareRemoteManagedHome: options.prepareRemoteManagedHome }
|
||||
|
|
@ -184,10 +185,13 @@ async function runExecutor(
|
|||
onMeta: async (payload: unknown) => {
|
||||
meta.push(payload as Record<string, unknown>);
|
||||
},
|
||||
onEvent: async (event: { eventType: string; payload?: Record<string, unknown> }) => {
|
||||
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<string, unknown> | 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<string, unknown> }>) {
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, string>;
|
||||
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<string, unknown>; 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<AcpxPreparedRuntime> {
|
||||
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<void>) | null = null;
|
||||
let freshDispose: (() => Promise<void>) | 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<void>) | null;
|
||||
dispose: (() => Promise<void>) | 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<string, string> = {};
|
||||
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<string, string> = {};
|
||||
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) {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<T>(
|
||||
ctx: Pick<AdapterExecutionContext, "onEvent">,
|
||||
now: () => number,
|
||||
step: string,
|
||||
fn: () => Promise<T>,
|
||||
): Promise<T> {
|
||||
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.
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue