mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
feat(native): split local runner startup into stage, launch and session-load spans (#19852)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - When a person messages an agent, the agent must start its turn fast. Time to first token (TTFT) and agent boot are first-class metrics. > - The Phase 0 baseline (#19846) found that local runner startup costs 1.6–17 s at p50 and up to 50 s at p95. It is the largest part of TTFT. > - The run log records that startup as one span (`runner.session.bootstrap` or `runner.session.resume`). We cannot see if the time goes to staging, to the `paperclip-runnerd` process launch, or to the provider session load. > - This pull request splits local runner startup into three sub-spans in the run log. > - The benefit: the follow-up work to cut follow-up-turn TTFT can measure each step before and after. ## Linked Issues or Issue Description Refs #19846 (TTFT baseline, finding 1 and gap "No sub-spans exist inside local runner startup"). Refs #19838. **What existing behavior does this improve?** The run-log performance spans for local native runner startup. **Current behavior** The local path records one `runner.session.bootstrap` or `runner.session.resume` span that covers runtime staging, the runner process launch and the provider handshake together. The remote path already records `runner.runtime.stage` and `runner.process.launch`; the local path does not. **Proposed behavior** The local path records three sequential child spans of `runner.session.startup`: `runner.runtime.stage`, `runner.process.launch` and `provider.session.load`. **Reason and benefit** The TTFT work needs to know which step dominates before it changes the runner lifecycle. Without these spans, a change to warm runner reuse cannot show which cost it removes. ## What Changed - `RunnerdCodexTransport` has a new optional `onStartupStep` callback. `#start` and `#resume` time three milestones and call it synchronously. A failing callback never changes startup. A nested startup (exact adoption from `thread/start`) keeps the outer clock. - `runner.runtime.stage`: launch preparation before the runner process exists. - `runner.process.launch`: spawn until runnerd accepts the run (`run.prepare`, or the recovery attachment on resume). An adopted live runner records no launch. This keeps the existing meaning in `summarizeStartupReuse`: a launch span means a fresh runner. - `provider.session.load`: `session.open` (or the recovery probe) until the provider identity is confirmed. - `recordNativeRunnerStartupStep` (`native-run-trace.ts`) records each step as a `run.performance.span` row with `mode` and `target: "local"`. It uses `trace.defer`, so the run-log write never delays `turn.start`. - The native executor passes the callback only for a local target. Remote launchers keep their own spans, so there are no duplicates. - `redaction.ts`: allowlist `runner.process.launch` and `provider.session.load` as native span names. - `doc/run-log-events.md`: new section "Local runner startup sub-steps". Data path (AGENTS.md §5.7): run log only, plus Observability spans when an operator sets an OTLP endpoint. The attributes (`mode`, `target`, `outcome`) are already in the closed span-attribute allowlist. No Telemetry change. ## Verification - `pnpm --filter @paperclipai/paperclip-runner exec vitest run src/live/runnerd-codex-transport-startup-steps.test.ts`: a real runnerd and the fake Codex fixture. Bootstrap and a cold resume each report stage, launch and session load in order, with `ok` outcomes and no overlap. A throwing sink does not change startup. - `cd server && npx vitest run src/services/native-runtime/native-run-trace.test.ts src/__tests__/redaction.test.ts`: the rows have the correct parent, offsets, durations and failed level. A stalled run-log sink does not block the caller. The rows settle before `task.provider_session.measured`. The names survive redaction. - The existing `runnerd-codex-transport*` suites pass (412 tests). Under host load (load average 16–27), the first full run had 7 timing failures in suites that do not set the callback. Each failing file passed when I ran it again alone. `runnerd-codex-transport-settlement` passed 3/3 for the failing case alone. - `tsc --noEmit` passes for `packages/paperclip-runner` and `server`. ## Risks - Low risk. Without `onStartupStep`, the transport behaves as before: the clock is not created and each milestone call is an optional-chain no-op. - More run-log rows: three per local runner startup. - The step boundaries are controller-side observations. `runner.process.launch` includes the runnerd authentication and the `run.prepare` acknowledgement, not only the OS spawn. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, file edit, test execution). ## 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 - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
e8a418cc4d
commit
a9ac2e2039
11 files changed
+289
-2
No files matched your search
@@ -99,6 +99,30 @@ and `harness_state.launch_material` for remote state inspection and fresh launch
|
||||
material. These events retain the existing schema and carry no paths or exception strings.
|
||||
They also become native Observability spans only with an operator OTLP endpoint.
|
||||
|
||||
## Local runner startup sub-steps
|
||||
|
||||
`run.performance.span` splits a local Runner startup (`runner.session.bootstrap` or
|
||||
`runner.session.resume`) into three sequential children of `runner.session.startup`:
|
||||
|
||||
- `runner.runtime.stage`: controller-side launch preparation before the Runner
|
||||
process exists (runtime context, provider home, launch profiles; on resume, the
|
||||
durable-state checks and authority rotation as well).
|
||||
- `runner.process.launch`: from spawning `paperclip-runnerd` until its first
|
||||
authenticated connection to the PRP control plane. An exact adoption of a live
|
||||
Runner records no launch, so this span keeps meaning a fresh Runner process
|
||||
(see startup reuse above).
|
||||
- `provider.session.load`: from that authentication until the provider session
|
||||
identity is confirmed. It covers `run.prepare` and `session.open`, or on resume
|
||||
the `run.attach` that restores the provider and the recovery probe.
|
||||
|
||||
Each carries `mode` (`bootstrap` or `resume`), `target: "local"`, and `outcome`
|
||||
(`failed` when startup fails inside the step). The Runner transport reports each
|
||||
milestone synchronously; the controller writes the row as deferred diagnostic
|
||||
work, so the run-log write never delays `turn.start`. Remote launchers keep their
|
||||
own `runner.runtime.stage` and `runner.process.launch` spans. These rows stay in
|
||||
the instance run log; with an operator OTLP endpoint they are also Observability
|
||||
spans. They are not Telemetry.
|
||||
|
||||
## Native PRP Run-Log Events
|
||||
|
||||
Provider startup command failures and failed provider turns can carry a
|
||||
|
||||
@@ -1344,6 +1344,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
}
|
||||
}
|
||||
"thread/resume" => {
|
||||
if let Some(delay) = argument(&args, "--thread-resume-delay-ms") {
|
||||
thread::sleep(Duration::from_millis(delay.parse()?));
|
||||
}
|
||||
if reject_missing_resume_state {
|
||||
send(json!({"id": id, "error": {
|
||||
"code": -32600,
|
||||
|
||||
@@ -305,6 +305,8 @@ export interface DurablePrpControlPlaneOptions {
|
||||
onProtocolIntegrityError?: (
|
||||
error: NativeSessionProtocolIntegrityError,
|
||||
) => void;
|
||||
/** Diagnostic only: called synchronously after each runner connection is authenticated. */
|
||||
onAuthenticatedConnection?: () => void;
|
||||
connectionLeaseTtlMs?: number;
|
||||
}
|
||||
|
||||
@@ -2035,6 +2037,7 @@ export class DurablePrpControlPlane {
|
||||
#recoverSemanticToolInput?: DurablePrpControlPlaneOptions["recoverSemanticToolInput"];
|
||||
#beforeAuthenticatedConnection?: DurablePrpControlPlaneOptions["beforeAuthenticatedConnection"];
|
||||
#onProtocolIntegrityError?: DurablePrpControlPlaneOptions["onProtocolIntegrityError"];
|
||||
#onAuthenticatedConnection?: DurablePrpControlPlaneOptions["onAuthenticatedConnection"];
|
||||
#protocolIntegrityError: NativeSessionProtocolIntegrityError | null = null;
|
||||
#connectionLeaseTtlMs: number;
|
||||
#persistCheckpoint: ((checkpoint: DurablePrpCheckpoint) => Promise<void>) | null = null;
|
||||
@@ -2093,6 +2096,7 @@ export class DurablePrpControlPlane {
|
||||
this.#recoverSemanticToolInput = options.recoverSemanticToolInput;
|
||||
this.#beforeAuthenticatedConnection = options.beforeAuthenticatedConnection;
|
||||
this.#onProtocolIntegrityError = options.onProtocolIntegrityError;
|
||||
this.#onAuthenticatedConnection = options.onAuthenticatedConnection;
|
||||
this.#connectionLeaseTtlMs = options.connectionLeaseTtlMs ?? 60_000;
|
||||
}
|
||||
|
||||
@@ -3773,6 +3777,7 @@ export class DurablePrpControlPlane {
|
||||
}
|
||||
|
||||
this.#store.state.connectionCount += 1;
|
||||
try { this.#onAuthenticatedConnection?.(); } catch { /* Diagnostics never change admission. */ }
|
||||
this.#connectionLeaseFailure = null;
|
||||
this.#store.state.lastLeaseId = lease.leaseId;
|
||||
this.#store.state.lastLeaseExpiresAt = lease.expiresAt;
|
||||
|
||||
@@ -97,6 +97,7 @@ export {
|
||||
type RetainedRunnerdMaintenanceEpochReceipt,
|
||||
type RunnerdCodexTransport,
|
||||
type RunnerdCodexTransportOptions,
|
||||
type RunnerdStartupStep,
|
||||
} from "./live/runnerd-codex-transport.js";
|
||||
export * from "./live/workspace-file-reference.js";
|
||||
export { readLinuxProcessStartedAt, type LinuxProcessStartOptions } from "./live/linux-process-start.js";
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { expect, it } from "vitest";
|
||||
import { createCapabilityRunnerdCodexTransport, type RunnerdStartupStep } from "./runnerd-codex-transport.js";
|
||||
import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js";
|
||||
|
||||
const STEP_ORDER = ["runner.runtime.stage", "runner.process.launch", "provider.session.load"];
|
||||
const PROVIDER_DELAY_MS = 1_500;
|
||||
|
||||
/** A slow provider must show up as session load, never as runner process launch. */
|
||||
function expectProviderTimeInSessionLoad(steps: RunnerdStartupStep[]): void {
|
||||
const duration = (name: string) => {
|
||||
const step = steps.find((candidate) => candidate.name === name)!;
|
||||
return step.endedAtMs - step.startedAtMs;
|
||||
};
|
||||
expect(duration("provider.session.load")).toBeGreaterThanOrEqual(PROVIDER_DELAY_MS);
|
||||
expect(duration("runner.process.launch")).toBeLessThan(PROVIDER_DELAY_MS);
|
||||
}
|
||||
|
||||
function expectOrderedSteps(steps: RunnerdStartupStep[], mode: RunnerdStartupStep["mode"]): void {
|
||||
expect(steps.map((step) => step.name)).toEqual(STEP_ORDER);
|
||||
for (const [index, step] of steps.entries()) {
|
||||
expect(step).toMatchObject({ mode, outcome: "ok" });
|
||||
expect(step.endedAtMs).toBeGreaterThanOrEqual(step.startedAtMs);
|
||||
// Sub-steps are sequential: each starts at or after the previous one ended.
|
||||
if (index > 0) expect(step.startedAtMs).toBeGreaterThanOrEqual(steps[index - 1]!.endedAtMs);
|
||||
}
|
||||
}
|
||||
|
||||
it("reports local stage, launch and provider session load for bootstrap and cold resume", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-startup-steps-"));
|
||||
const identity = {
|
||||
runnerInstanceId: "startup-steps-runner",
|
||||
environmentLeaseId: "startup-steps-lease",
|
||||
runId: "first-run",
|
||||
normalizedSessionId: "startup-steps-session",
|
||||
turnId: "first-turn",
|
||||
itemId: "first-item",
|
||||
};
|
||||
const options = {
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(),
|
||||
codexCommand: fakeCodex,
|
||||
// Delay the provider's own session start and resume. The runner process
|
||||
// authenticates first, so the delay belongs to provider.session.load.
|
||||
codexArgs: fakeCodexArgs(root, "--thread-start-delay-ms", String(PROVIDER_DELAY_MS),
|
||||
"--thread-resume-delay-ms", String(PROVIDER_DELAY_MS)),
|
||||
stateDirectory: root,
|
||||
lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 },
|
||||
runnerReconnectGraceMs: 2_000,
|
||||
};
|
||||
const bootstrapSteps: RunnerdStartupStep[] = [];
|
||||
const first = createCapabilityRunnerdCodexTransport({
|
||||
...options, prpIdentity: identity, onStartupStep: (step) => { bootstrapSteps.push(step); },
|
||||
});
|
||||
let resumed: ReturnType<typeof createCapabilityRunnerdCodexTransport> | undefined;
|
||||
try {
|
||||
const opened = await first.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] });
|
||||
expectOrderedSteps(bootstrapSteps, "bootstrap");
|
||||
expectProviderTimeInSessionLoad(bootstrapSteps);
|
||||
const thread = opened.thread as { id: string; sessionId: string };
|
||||
await first.transport.close();
|
||||
// Closing a live transport is not a startup: no further steps.
|
||||
expect(bootstrapSteps).toHaveLength(3);
|
||||
|
||||
const resumeSteps: RunnerdStartupStep[] = [];
|
||||
// A new run on the saved session rotates authority through run.attach,
|
||||
// which restores the provider before the command completes.
|
||||
resumed = createCapabilityRunnerdCodexTransport({
|
||||
...options,
|
||||
prpIdentity: { ...identity, runId: "second-run", turnId: "second-turn", itemId: "second-item" },
|
||||
resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId },
|
||||
onStartupStep: (step) => { resumeSteps.push(step); },
|
||||
});
|
||||
await resumed.transport.request("thread/read", {});
|
||||
expectOrderedSteps(resumeSteps, "resume");
|
||||
expectProviderTimeInSessionLoad(resumeSteps);
|
||||
} finally {
|
||||
await resumed?.transport.close().catch(() => undefined);
|
||||
await first.transport.close().catch(() => undefined);
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
}, 60_000);
|
||||
|
||||
it("never lets a failing startup-step sink change startup", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-startup-steps-sink-"));
|
||||
const names: string[] = [];
|
||||
const bundle = createCapabilityRunnerdCodexTransport({
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(),
|
||||
codexCommand: fakeCodex,
|
||||
codexArgs: fakeCodexArgs(root),
|
||||
stateDirectory: root,
|
||||
onStartupStep: (step) => {
|
||||
names.push(step.name);
|
||||
throw new Error("diagnostic sink unavailable");
|
||||
},
|
||||
});
|
||||
try {
|
||||
const opened = await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] });
|
||||
expect((opened.thread as { id?: string }).id).toEqual(expect.any(String));
|
||||
expect(names).toEqual(STEP_ORDER);
|
||||
} finally {
|
||||
await bundle.transport.close().catch(() => undefined);
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
}, 60_000);
|
||||
@@ -1569,6 +1569,11 @@ export interface CapabilityRunnerdCodexTransportOptions extends RunnerExecutionD
|
||||
/** Controller-owned opt-in operation timing. No exporter lives in Runner. */
|
||||
measureOperation?<T>(phase: string, work: () => Promise<T>, attributes?: Record<string, number | boolean>): Promise<T>;
|
||||
recordTimingReceipt?(receipt: unknown): void;
|
||||
/**
|
||||
* Controller-owned startup sub-steps for the run log. Called synchronously
|
||||
* at each milestone; the controller must not wait on it before turn.start.
|
||||
*/
|
||||
onStartupStep?(step: RunnerdStartupStep): void;
|
||||
onDiagnostic?: (message: string) => void;
|
||||
onEvidence?: (evidence: Readonly<CapabilityRunnerdProcessEvidence>) => void;
|
||||
/** Persist process ownership immediately after spawn, before waiting for
|
||||
@@ -1636,6 +1641,15 @@ export interface CapabilityRunnerdCodexTransportOptions extends RunnerExecutionD
|
||||
}) => void | Promise<void>;
|
||||
}
|
||||
|
||||
/** One measured local startup sub-step, named for the run-log span it becomes. */
|
||||
export interface RunnerdStartupStep {
|
||||
name: "runner.runtime.stage" | "runner.process.launch" | "provider.session.load";
|
||||
mode: "bootstrap" | "resume";
|
||||
startedAtMs: number;
|
||||
endedAtMs: number;
|
||||
outcome: "ok" | "failed";
|
||||
}
|
||||
|
||||
export type RunnerdCodexTransportOptions =
|
||||
CapabilityRunnerdCodexTransportOptions;
|
||||
|
||||
@@ -3950,6 +3964,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
#nextTraceDebugSequence = 1;
|
||||
#traceRehydrationSpoolOverflow = false;
|
||||
readonly #traceFrameIndex = new RunnerdTraceFrameIndex();
|
||||
#startupSteps: {
|
||||
begin(name: RunnerdStartupStep["name"]): void;
|
||||
/** Move on only from the named open step (later reconnects are not startup milestones). */
|
||||
advance(from: RunnerdStartupStep["name"], to: RunnerdStartupStep["name"]): void;
|
||||
end(): void;
|
||||
} | null = null;
|
||||
#pendingTraceRehydrations: PendingTraceRehydration[] = [];
|
||||
#pendingDriverTraceInterpretations: PendingDriverTraceInterpretation[] = [];
|
||||
#restoredRuntimeRequests: CodexRpcServerRequest[] = [];
|
||||
@@ -5781,7 +5801,43 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
async #start(
|
||||
params: Record<string, unknown>,
|
||||
): Promise<Record<string, unknown>> {
|
||||
return this.options.measureOperation ? this.options.measureOperation("runner.transport.cold_start", () => this.#startWork(params)) : this.#startWork(params);
|
||||
const work = () => this.#withStartupSteps("bootstrap", () => this.#startWork(params));
|
||||
return this.options.measureOperation ? this.options.measureOperation("runner.transport.cold_start", work) : work();
|
||||
}
|
||||
|
||||
/**
|
||||
* Time this startup's sub-steps for the controller's run log. A nested
|
||||
* startup (exact adoption from thread/start) keeps the outer clock.
|
||||
*/
|
||||
async #withStartupSteps<T>(mode: RunnerdStartupStep["mode"], work: () => Promise<T>): Promise<T> {
|
||||
const onStep = this.options.onStartupStep;
|
||||
if (!onStep || this.#startupSteps !== null) return work();
|
||||
let open: { name: RunnerdStartupStep["name"]; startedAtMs: number } | null = null;
|
||||
const close = (outcome: RunnerdStartupStep["outcome"]) => {
|
||||
if (!open) return;
|
||||
const step = { ...open, mode, endedAtMs: Date.now(), outcome };
|
||||
open = null;
|
||||
try { onStep(step); } catch { /* Diagnostics never change startup. */ }
|
||||
};
|
||||
this.#startupSteps = {
|
||||
begin: name => { close("ok"); open = { name, startedAtMs: Date.now() }; },
|
||||
advance: (from, to) => {
|
||||
if (open?.name !== from) return;
|
||||
close("ok");
|
||||
open = { name: to, startedAtMs: Date.now() };
|
||||
},
|
||||
end: () => close("ok"),
|
||||
};
|
||||
try {
|
||||
const value = await work();
|
||||
close("ok");
|
||||
return value;
|
||||
} catch (error) {
|
||||
close("failed");
|
||||
throw error;
|
||||
} finally {
|
||||
this.#startupSteps = null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Runner state held by an external Runner host (a plugin task or a sandbox). */
|
||||
@@ -5838,6 +5894,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
expectedRunnerVersion: runnerArtifact.version,
|
||||
expectedRunnerDigest: runnerArtifact.digest,
|
||||
onProtocolIntegrityError: (error) => this.#failTransport(error),
|
||||
// The runner process is up once it authenticates; provider work follows.
|
||||
onAuthenticatedConnection: () => this.#startupSteps?.advance("runner.process.launch", "provider.session.load"),
|
||||
get authorityRunId() { return authorityRunId(); },
|
||||
semanticInputMode: this.#retainsSourceInputs() || this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch",
|
||||
onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call),
|
||||
@@ -5866,7 +5924,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
});
|
||||
}
|
||||
const unclaimedBootstrap = core.unclaimedBootstrapCheckpoint();
|
||||
this.#startupSteps?.begin("runner.runtime.stage");
|
||||
const prepared = await this.#prepareLaunch(params, identity, dynamicTools);
|
||||
this.#startupSteps?.end();
|
||||
const { runAttachTemplate } = prepared;
|
||||
// Preserve the first generation's provider attachment seed independently
|
||||
// of bounded command history. The in-memory copy serves a live warm
|
||||
@@ -5897,6 +5957,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
const bootstrapTicket = core.issueBootstrapTicket(RUNNER_BOOTSTRAP_TICKET_TTL_MS);
|
||||
await core.flushCheckpoint();
|
||||
this.#assertStartupOpen();
|
||||
this.#startupSteps?.begin("runner.process.launch");
|
||||
const handle = spawnRunner({
|
||||
connection: registration?.connection ?? {
|
||||
mode: "connect",
|
||||
@@ -5942,8 +6003,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#publish();
|
||||
this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
|
||||
await this.#waitCommand("run.prepare", undefined, undefined, true);
|
||||
this.#startupSteps?.advance("runner.process.launch", "provider.session.load");
|
||||
await this.#waitCommand("session.open", undefined, admissionDeadline, true);
|
||||
await this.#waitForProviderIdentity(undefined, admissionDeadline);
|
||||
this.#startupSteps?.end();
|
||||
this.#startupComplete = true;
|
||||
this.#diagnostic("runnerd authenticated to the durable PRP control plane");
|
||||
return this.#openedThreadResponse(params);
|
||||
@@ -5988,11 +6051,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
|
||||
async #resume(): Promise<void> {
|
||||
return this.options.measureOperation ? this.options.measureOperation("runner.transport.resume", () => this.#resumeWork()) : this.#resumeWork();
|
||||
const work = () => this.#withStartupSteps("resume", () => this.#resumeWork());
|
||||
return this.options.measureOperation ? this.options.measureOperation("runner.transport.resume", work) : work();
|
||||
}
|
||||
|
||||
async #resumeWork(): Promise<void> {
|
||||
if (this.options.acpxCandidateAdmission?.agent === "antigravity") throw new Error("ACPX candidate durable runner recovery is not admitted");
|
||||
this.#startupSteps?.begin("runner.runtime.stage");
|
||||
let desiredIdentity = this.options.prpIdentity;
|
||||
if (desiredIdentity === undefined) {
|
||||
throw new Error("PRP provider resume state is unavailable");
|
||||
@@ -6350,6 +6415,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
expectedRunnerVersion: runnerArtifact.version,
|
||||
expectedRunnerDigest: runnerArtifact.digest,
|
||||
onProtocolIntegrityError: (error) => this.#failTransport(error),
|
||||
// The runner process is up once it authenticates; provider work follows.
|
||||
onAuthenticatedConnection: () => this.#startupSteps?.advance("runner.process.launch", "provider.session.load"),
|
||||
get authorityRunId() { return authorityRunId(); },
|
||||
semanticInputMode: this.#retainsSourceInputs() || this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch",
|
||||
onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call),
|
||||
@@ -6709,6 +6776,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
const admissionDeadline = this.#sourceAttachmentDeadline === null
|
||||
? this.#coldAdmissionDeadline(adoptedRunner === undefined)
|
||||
: Math.min(Date.now() + 30_000, this.#sourceAttachmentDeadline);
|
||||
// An adopted runner is already live: no launch step (a launch span means a fresh process).
|
||||
if (adoptedRunner) this.#startupSteps?.end();
|
||||
else this.#startupSteps?.begin("runner.process.launch");
|
||||
const handle = adoptedRunner
|
||||
? null
|
||||
: spawnRunner({
|
||||
@@ -6860,6 +6930,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
if (runAttachment) {
|
||||
await this.#waitCommand("run.attach", runAttachment.commandId, admissionDeadline, true);
|
||||
}
|
||||
// A launched runner normally advanced at authentication, before run.attach
|
||||
// restores the provider. An adopted runner had no launch step.
|
||||
if (adoptedRunner) this.#startupSteps?.begin("provider.session.load");
|
||||
else this.#startupSteps?.advance("runner.process.launch", "provider.session.load");
|
||||
if (recoveryProbeCommandId !== null) {
|
||||
await this.#waitCommand(sourceAttachmentTarget ? "session.open" : "runner.drain", recoveryProbeCommandId, admissionDeadline, true);
|
||||
if (this.#checkpointProviderIdentityExpectation !== null) {
|
||||
@@ -6889,6 +6963,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
: undefined,
|
||||
admissionDeadline,
|
||||
);
|
||||
this.#startupSteps?.end();
|
||||
this.#startupComplete = true;
|
||||
if (recoveringAttachment && staged) {
|
||||
this.#sourceRecoveryIdentity = null;
|
||||
|
||||
@@ -399,6 +399,9 @@ describe("redaction", () => {
|
||||
"runner.prp.route.register",
|
||||
"runner.transport.connect",
|
||||
"runner.session.bootstrap",
|
||||
"runner.runtime.stage",
|
||||
"runner.process.launch",
|
||||
"provider.session.load",
|
||||
"runner.turn.submit",
|
||||
"runner.session.startup",
|
||||
"provider_pack.select",
|
||||
|
||||
@@ -334,6 +334,7 @@ const NATIVE_RUN_SPAN_NAMES = new Set([
|
||||
"native.result.finalize",
|
||||
"native.session.execute",
|
||||
"provider.session.continuity_break",
|
||||
"provider.session.load",
|
||||
"provider.session.resume",
|
||||
"provider.time_to_first_agent_event",
|
||||
"provider.turn.queue",
|
||||
@@ -352,6 +353,7 @@ const NATIVE_RUN_SPAN_NAMES = new Set([
|
||||
"runner.artifact.discover",
|
||||
"runner.artifact.prepare",
|
||||
"runner.prp.authenticate",
|
||||
"runner.process.launch",
|
||||
"runner.prp.route.register",
|
||||
"runner.runtime.stage",
|
||||
"runner.session.bootstrap",
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
nativeRunPreparationStarts,
|
||||
recordFailedSkillPreparation,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
recordNativeRunnerStartupStep,
|
||||
resolveNativeTimeToFirstTokenStart,
|
||||
NATIVE_RUN_SPAN_EVENT_TYPE,
|
||||
NATIVE_RUN_TRACE_SCHEMA_VERSION,
|
||||
@@ -604,3 +605,41 @@ describe("deferred diagnostics", () => {
|
||||
expect(late).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("local runner startup sub-steps", () => {
|
||||
it("records each step beneath runner.session.startup without blocking the caller", async () => {
|
||||
const events: AdapterRuntimeEvent[] = [];
|
||||
let releaseSink!: () => void;
|
||||
const sinkGate = new Promise<void>((resolve) => { releaseSink = resolve; });
|
||||
const { traceContext, spans } = createRecordingTraceContext();
|
||||
const trace = createNativeRunTrace({
|
||||
runId: "startup-steps", startedAtMs: 1_000, traceContext,
|
||||
onEvent: async (event) => { await sinkGate; events.push(event); },
|
||||
});
|
||||
const startup = trace.start("runner.session.startup", { parentName: "task.provider_session", startedAtMs: 1_000 });
|
||||
// Synchronous and detached: a stalled run-log sink cannot delay turn.start.
|
||||
recordNativeRunnerStartupStep(trace, { name: "runner.runtime.stage", mode: "bootstrap", startedAtMs: 1_000, endedAtMs: 1_200, outcome: "ok" });
|
||||
recordNativeRunnerStartupStep(trace, { name: "runner.process.launch", mode: "bootstrap", startedAtMs: 1_200, endedAtMs: 1_900, outcome: "ok" });
|
||||
recordNativeRunnerStartupStep(trace, { name: "provider.session.load", mode: "resume", startedAtMs: 1_900, endedAtMs: 4_000, outcome: "failed" });
|
||||
expect(events).toHaveLength(0);
|
||||
const ended = trace.end(startup, { endedAtMs: 4_000 });
|
||||
const finished = trace.finish("ok");
|
||||
releaseSink();
|
||||
await ended;
|
||||
await finished;
|
||||
const payloads = events.map((event) => event.payload as Record<string, unknown>);
|
||||
const step = (name: string) => payloads.find((payload) => payload.span === name);
|
||||
expect(step("runner.runtime.stage")).toMatchObject({ parentSpan: "runner.session.startup", startOffsetMs: 0, durationMs: 200, outcome: "ok", mode: "bootstrap", target: "local" });
|
||||
expect(step("runner.process.launch")).toMatchObject({ parentSpan: "runner.session.startup", startOffsetMs: 200, durationMs: 700, outcome: "ok" });
|
||||
expect(step("provider.session.load")).toMatchObject({ parentSpan: "runner.session.startup", durationMs: 2_100, outcome: "failed", mode: "resume" });
|
||||
expect(events.find((event) => (event.payload as Record<string, unknown>).span === "provider.session.load")?.level).toBe("warn");
|
||||
expect(spans.find((span) => span.name === "runner.process.launch")?.parentName).toBe("runner.session.startup");
|
||||
// Settled before the root measurement closes.
|
||||
const names = payloads.map((payload) => payload.span);
|
||||
expect(names.indexOf("provider.session.load")).toBeLessThan(names.indexOf("task.provider_session.measured"));
|
||||
});
|
||||
|
||||
it("is a no-op without a trace", () => {
|
||||
expect(() => recordNativeRunnerStartupStep(undefined, { name: "runner.process.launch", mode: "bootstrap", startedAtMs: 1, endedAtMs: 2, outcome: "ok" })).not.toThrow();
|
||||
});
|
||||
});
|
||||
@@ -625,6 +625,32 @@ export async function recordNativeProviderPerformanceSpans(trace: NativeRunTrace
|
||||
} catch { /* Diagnostics cannot replace the work's result. */ }
|
||||
}
|
||||
|
||||
/** One local runner startup sub-step measured by the runner transport. */
|
||||
export interface NativeRunnerStartupStep {
|
||||
name: "runner.runtime.stage" | "runner.process.launch" | "provider.session.load";
|
||||
mode: "bootstrap" | "resume";
|
||||
startedAtMs: number;
|
||||
endedAtMs: number;
|
||||
outcome: NativeRunSpanOutcome;
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a local runner startup sub-step under `runner.session.startup`.
|
||||
* Run log (and Observability when an operator endpoint is set) only, never
|
||||
* Telemetry. Detached: the run-log write never delays `turn.start`; the
|
||||
* trace settles it at finish.
|
||||
*/
|
||||
export function recordNativeRunnerStartupStep(trace: NativeRunTrace | undefined, step: NativeRunnerStartupStep): void {
|
||||
trace?.defer(() => trace.record({
|
||||
name: step.name,
|
||||
parentName: "runner.session.startup",
|
||||
startedAtMs: step.startedAtMs,
|
||||
endedAtMs: step.endedAtMs,
|
||||
outcome: step.outcome,
|
||||
attributes: { mode: step.mode, target: "local" },
|
||||
}));
|
||||
}
|
||||
|
||||
/** Emit preparation failure even when execution aborts before a native session exists. */
|
||||
export async function recordFailedSkillPreparation(input: {
|
||||
runId: string;
|
||||
|
||||
@@ -284,6 +284,7 @@ import {
|
||||
isNativeRunRootHistoricalSpan,
|
||||
nativeRunPreparationStarts,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
recordNativeRunnerStartupStep,
|
||||
resolveNativeTimeToFirstTokenStart,
|
||||
type NativeRunHistoricalSpan,
|
||||
type NativeRunSpanScope,
|
||||
@@ -16003,6 +16004,8 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
...(input.recoveryPurpose === "stop_cleanup" ? { recoveryPurpose: "stop_cleanup" as const } : {}),
|
||||
measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined,
|
||||
recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined,
|
||||
// Remote launchers record their own stage and launch spans.
|
||||
...(remoteTarget ? {} : { onStartupStep: (step: Parameters<typeof recordNativeRunnerStartupStep>[1]) => recordNativeRunnerStartupStep(input.trace, step) }),
|
||||
onSpawn: input.onSpawn,
|
||||
...(input.execution.provider.kind === "codex" ? { disableCodexWebSearch: "webSearch" in input.execution.provider && input.execution.provider.webSearch === "disabled" } : {}),
|
||||
homeMode: preconfiguredHome ? "preconfigured" : "isolated",
|
||||
|
||||
Reference in new issue
Block a user