mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
perf(run-log): record named preparation phases before dispatch as run-log spans (#20046)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Every native run records performance spans in its run log, the instance-local `heartbeat_run_events` table, so slow turns can be diagnosed without any external service > - The steps between environment acquisition and dispatch are timed by named preparation phases, but those timings reach only OpenTelemetry, which is off unless an operator configures an endpoint > - On a Daytona follow-up in a paperclip-sized workspace, 17–30 s sat in one opaque span, `heartbeat.prepare_after_environment` > - This pull request also records the named preparation phases, and their nested steps, as run-log spans > - The benefit is that the next slow step shows up by name in the run log; this change found a full workspace re-upload on every remote turn ## Linked Issues or Issue Description **What happened?** The run log showed `heartbeat.prepare_after_environment` at 17–30 s on Daytona follow-ups, with nothing inside it. The phases that would explain it (`heartbeat.prepare.*`, `runtime_selection.*`, `workspace_sync.*`) were recorded only as OpenTelemetry spans, and those are a no-op without an OTLP endpoint. **Steps to reproduce** 1. Run a Codex task on a Daytona environment with a large repository as the project workspace. 2. Post a follow-up comment. 3. The run log shows one large `heartbeat.prepare_after_environment` span with no children. **Expected behavior** The run log names the steps that make up the wait. ## What Changed - `execution-preparation-timing.ts`: every named preparation phase (`measure`) and nested run phase (`measureRunPhase`) records its start, end and outcome in a per-run in-memory list. - Bounds: phases under 50 ms are skipped, at most 200 per run, at most 256 runs held. - Without a tracer, the run scope is still set, so nested phases reach the run. - OpenTelemetry behavior is unchanged. - `runtime-execution.ts`: at native dispatch, the recorded phases are drained into the run's preparation spans under `task.prepare`. - `runtime-selection.ts`: the remote Codex CLI version read and the runtime context build get their own phases, `runtime_selection.codex_cli_version` and `runtime_selection.runtime_context`. ## Verification - `execution-preparation-phases.test.ts`: a slow phase is recorded and a fast one skipped, a failure is marked, draining clears the run, runs are kept apart, and a nested `measureRunPhase` step is recorded without a tracer. - `database-operation-attribution`, `company-skills-observability` and `github-catalog-timing`: pass (15 tests in total). - Daytona Codex run in a fresh paperclip clone: the follow-up's 17.8 s gap now reads `heartbeat.prepare.runtime_selection` 15.7 s, then `runtime_selection.workspace_sync` 15.4 s, then `workspace_sync.transfer` 13.3 s. ## Risks - This is a run-log change only, not Telemetry: the data stays in the instance database. It adds roughly 10–40 spans per run, bounded as above. - The recorder is in memory and per process. A run that never dispatches is evicted when the cap of 256 runs is reached. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file editing, test execution, Daytona measurements). ## 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
96badeaa42
commit
cf9c81d9a7
4 files changed
+148
-11
No files matched your search
@@ -0,0 +1,63 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { createRunPreparationTiming, drainRunPreparationPhases, measureRunPhase, withRunExecutionTiming } from "../services/execution-preparation-timing.js";
|
||||
|
||||
describe("run preparation phases for the run log", () => {
|
||||
afterEach(() => { vi.unstubAllEnvs(); });
|
||||
it("records slow named phases per run, marks failures, and forgets them once drained", async () => {
|
||||
const runId = randomUUID();
|
||||
const time = createRunPreparationTiming(runId);
|
||||
await time("workspace_provision", () => delay(70));
|
||||
await time("workspace_binding", async () => undefined);
|
||||
await expect(time("runner_binary", async () => { await delay(60); throw new Error("upload failed"); })).rejects.toThrow("upload failed");
|
||||
await Promise.resolve();
|
||||
const phases = drainRunPreparationPhases(runId);
|
||||
expect(phases.map(phase => [phase.name, phase.outcome])).toEqual([
|
||||
["heartbeat.prepare.workspace_provision", "ok"],
|
||||
["heartbeat.prepare.runner_binary", "failed"],
|
||||
]);
|
||||
expect(phases[0]!.endedAtMs - phases[0]!.startedAtMs).toBeGreaterThanOrEqual(50);
|
||||
expect(drainRunPreparationPhases(runId)).toEqual([]);
|
||||
});
|
||||
|
||||
it("records nested run phases inside a preparation phase without a tracer", async () => {
|
||||
const runId = randomUUID();
|
||||
await createRunPreparationTiming(runId)("runtime_selection", async () => {
|
||||
await measureRunPhase("runtime_selection.provider_discovery", () => delay(60));
|
||||
await measureRunPhase("runtime_selection.skill_assets", async () => undefined);
|
||||
});
|
||||
expect(drainRunPreparationPhases(runId).map(phase => phase.name)).toEqual([
|
||||
"runtime_selection.provider_discovery", "heartbeat.prepare.runtime_selection",
|
||||
]);
|
||||
});
|
||||
|
||||
it("never records an execution attempt, so a resumed run's prepare holds only its own preparation", async () => {
|
||||
const runId = randomUUID();
|
||||
await createRunPreparationTiming(runId)("workspace", () => delay(60));
|
||||
expect(drainRunPreparationPhases(runId)).toHaveLength(1);
|
||||
await withRunExecutionTiming(runId, async () => {
|
||||
await delay(60);
|
||||
await measureRunPhase("workspace_sync.transfer", () => delay(60));
|
||||
});
|
||||
expect(drainRunPreparationPhases(runId)).toEqual([]);
|
||||
});
|
||||
|
||||
it("records nested preparation steps when an OTLP endpoint is set too", async () => {
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://127.0.0.1:9");
|
||||
const runId = randomUUID();
|
||||
await createRunPreparationTiming(runId)("runtime_selection", async () => {
|
||||
await measureRunPhase("runtime_selection.workspace_sync", () => delay(60));
|
||||
});
|
||||
expect(drainRunPreparationPhases(runId).map(phase => phase.name)).toEqual([
|
||||
"runtime_selection.workspace_sync", "heartbeat.prepare.runtime_selection",
|
||||
]);
|
||||
});
|
||||
|
||||
it("keeps runs apart", async () => {
|
||||
const [first, second] = [randomUUID(), randomUUID()];
|
||||
await createRunPreparationTiming(first)("instructions", () => delay(60));
|
||||
expect(drainRunPreparationPhases(second)).toEqual([]);
|
||||
expect(drainRunPreparationPhases(first)).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
@@ -73,7 +73,7 @@ export type ExecutionPhase =
|
||||
| "workspace.checkout.cache_lookup" | "workspace.checkout.cache_publish"
|
||||
| "runtime_selection.review" | "runtime_selection.contract_read" | "runtime_selection.contract_ensure"
|
||||
| "runtime_selection.resume" | "runtime_selection.recovery_profile" | "runtime_selection.profile" | "runtime_selection.interactions"
|
||||
| "runtime_selection.provider_discovery" | "runtime_selection.bootstrap" | "runtime_selection.run_write" | "runtime_selection.finalization_write"
|
||||
| "runtime_selection.provider_discovery" | "runtime_selection.codex_cli_version" | "runtime_selection.runtime_context" | "runtime_selection.bootstrap" | "runtime_selection.run_write" | "runtime_selection.finalization_write"
|
||||
| "runtime_selection.transaction" | "runtime_selection.transaction.entry" | "runtime_selection.transaction.commit" | "runtime_selection.transaction.rollback"
|
||||
| "runtime_selection.row_lock" | "runtime_selection.workspace_sync" | "runtime_selection.skill_assets"
|
||||
| "runtime_context.instructions" | "runtime_context.skills" | "runtime_context.mcp"
|
||||
@@ -147,7 +147,55 @@ export function isRunPhaseActive(phase: ExecutionPhase): boolean {
|
||||
return endpointEnabled() && owned?.phase === phase && !owned.ended;
|
||||
}
|
||||
|
||||
/**
|
||||
* Named preparation phases per run for the run log, so a slow step before
|
||||
* dispatch is visible without an OTLP endpoint. Bounded: phases under 50 ms are
|
||||
* left out, at most 200 per run, and at most 256 runs are held until drained.
|
||||
*/
|
||||
export type RecordedPreparationPhase = { name: string; startedAtMs: number; endedAtMs: number; outcome: "ok" | "failed" };
|
||||
const recordedPreparationPhases = new Map<string, RecordedPreparationPhase[]>();
|
||||
const MIN_RECORDED_PHASE_MS = 50;
|
||||
const MAX_RECORDED_PHASES_PER_RUN = 200;
|
||||
const MAX_RECORDED_RUNS = 256;
|
||||
|
||||
function isPreparationPhaseName(name: string): boolean {
|
||||
return name.startsWith("heartbeat.prepare.");
|
||||
}
|
||||
|
||||
function recordPreparationPhase(runId: string, phase: RecordedPreparationPhase) {
|
||||
if (phase.endedAtMs - phase.startedAtMs < MIN_RECORDED_PHASE_MS) return;
|
||||
let phases = recordedPreparationPhases.get(runId);
|
||||
if (!phases) {
|
||||
if (recordedPreparationPhases.size >= MAX_RECORDED_RUNS) {
|
||||
const oldest = recordedPreparationPhases.keys().next().value;
|
||||
if (oldest !== undefined) recordedPreparationPhases.delete(oldest);
|
||||
}
|
||||
phases = [];
|
||||
recordedPreparationPhases.set(runId, phases);
|
||||
}
|
||||
if (phases.length < MAX_RECORDED_PHASES_PER_RUN) phases.push(phase);
|
||||
}
|
||||
|
||||
/** Returns and forgets the recorded preparation phases of a run. */
|
||||
export function drainRunPreparationPhases(runId: string): RecordedPreparationPhase[] {
|
||||
const phases = recordedPreparationPhases.get(runId) ?? [];
|
||||
recordedPreparationPhases.delete(runId);
|
||||
return phases;
|
||||
}
|
||||
|
||||
function measure<T>(runId: string, name: string, work: () => T | Promise<T>, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context): Promise<T> {
|
||||
const startedAtMs = Date.now();
|
||||
const operation = measureTraced(runId, name, work, independent, counts, activeParent);
|
||||
// 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(
|
||||
() => recordPreparationPhase(runId, { name, startedAtMs, endedAtMs: Date.now(), outcome: "ok" }),
|
||||
() => recordPreparationPhase(runId, { name, startedAtMs, endedAtMs: Date.now(), outcome: "failed" }),
|
||||
);
|
||||
return operation;
|
||||
}
|
||||
|
||||
function measureTraced<T>(runId: string, name: string, work: () => T | Promise<T>, independent = false, counts: { fileCount?: number; bytes?: number } = {}, activeParent?: Context): Promise<T> {
|
||||
let operation: Promise<T> | undefined;
|
||||
const run = (span?: { setStatus(status: { code: number }): void; end(): void }, parent?: Context) => {
|
||||
if (operation) return operation;
|
||||
@@ -170,7 +218,15 @@ function measure<T>(runId: string, name: string, work: () => T | Promise<T>, ind
|
||||
}));
|
||||
return operation;
|
||||
};
|
||||
if (!endpointEnabled()) return withDatabaseOperation(name, async () => work());
|
||||
if (!endpointEnabled()) {
|
||||
// No tracer: still scope the run so nested measureRunPhase steps reach the run log.
|
||||
const previous = ownership.getStore();
|
||||
const owned: RunScope = { runId, phase: name, parent: undefined, ended: false,
|
||||
details: !independent && previous?.runId === runId ? previous.details : new Map<string, number>() };
|
||||
return withDatabaseOperation(name, () => ownership.run(owned, async () => {
|
||||
try { return await work(); } finally { owned.ended = true; }
|
||||
}));
|
||||
}
|
||||
try {
|
||||
const parent = independent ? undefined : activeParent ?? getRunExecutionTraceContext(runId);
|
||||
const trigger = parent ? undefined : trace.getActiveSpan()?.spanContext();
|
||||
@@ -211,13 +267,25 @@ export function withRunResponsePresentationTiming<T>(runId: string, work: () =>
|
||||
export function measureRunPhase<T>(phase: ExecutionPhase, work: () => T | Promise<T>, counts?: { fileCount?: number; bytes?: number }): Promise<T> {
|
||||
return withDatabaseOperation(phase, async () => {
|
||||
const owned = ownership.getStore();
|
||||
if (!endpointEnabled() || !owned || owned.ended) return work();
|
||||
if (!owned || owned.ended) return work();
|
||||
const count = owned.details.get(phase) ?? 0;
|
||||
if (count >= 16) return work();
|
||||
owned.details.set(phase, count + 1);
|
||||
// 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.
|
||||
return measure(owned.runId, phase, work, false, counts, getActiveStepContext()?.parentContext as Context | undefined);
|
||||
// A step inside a preparation phase reaches the run log with or without a tracer.
|
||||
const recordedRunId = isPreparationPhaseName(owned.phase) ? owned.runId : 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 work();
|
||||
outcome = "ok";
|
||||
return value;
|
||||
} finally {
|
||||
if (recordedRunId) recordPreparationPhase(recordedRunId, { name: phase, startedAtMs, endedAtMs: Date.now(), outcome });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ import { environmentService } from "../environments.js";
|
||||
import { writeExecutionControl } from "../execution-control-notifications.js";
|
||||
import { inspectPluginWorkspaceForFinalization } from "../native-runtime/native-plugin-workspace-finalization.js";
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
import { measureRunPhase } from "../execution-preparation-timing.js";
|
||||
import { drainRunPreparationPhases, measureRunPhase } from "../execution-preparation-timing.js";
|
||||
import { parseNativeSessionGoalControl } from "./scheduling.js";
|
||||
import { isHeartbeatRunTerminalStatus } from "./run-lifecycle.js";
|
||||
import {
|
||||
@@ -601,6 +601,12 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
|
||||
});
|
||||
if (wakeIngressSpan)
|
||||
nativeRunnerPreparationSpans.unshift(wakeIngressSpan);
|
||||
// Named preparation phases (workspace, configuration, runner staging)
|
||||
// that ran before dispatch, as run-log spans under task.prepare.
|
||||
nativeRunnerPreparationSpans.push(...drainRunPreparationPhases(run.id).map(phase => ({
|
||||
name: phase.name, startedAtMs: phase.startedAtMs, endedAtMs: phase.endedAtMs,
|
||||
...(phase.outcome === "failed" ? { outcome: "failed" as const } : {}),
|
||||
})));
|
||||
nativeRunnerPreparationSpans.push(
|
||||
...buildNativeHeartbeatPreparationSpans({
|
||||
runCreatedAtMs,
|
||||
|
||||
@@ -733,14 +733,14 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele
|
||||
? await museRunnerBroker(db).snapshot(agent.companyId, agent.id, String(parseObject(agent.adapterConfig).museBindingId ?? "")) : undefined;
|
||||
const dotBinding = ["openai_dot_mcp", "grok_bot_mcp"].includes(nativeRuntimeResolution.profile.backend)
|
||||
? await dotRunnerBroker(db).snapshot(agent.companyId, agent.id, String(parseObject(agent.adapterConfig)[nativeRuntimeResolution.profile.backend === "grok_bot_mcp" ? "externalBindingId" : "dotBindingId"] ?? "")) : undefined;
|
||||
const nativeRuntimeContext = await buildNativeRuntimeContext({
|
||||
const nativeRuntimeContext = await measureRunPhase("runtime_selection.runtime_context", () => buildNativeRuntimeContext({
|
||||
db,
|
||||
agent,
|
||||
runId: run.id,
|
||||
runtimeConfig,
|
||||
runtimeSkillEntries,
|
||||
instructionWorkingCopy: instructionCopy ? { rootPath: instructionCopy.executionRoot, entryPath: instructionCopy.entryFile, ...(isAgentDirectoryCopy(instructionCopy) ? { kind: "agent_files" as const } : {}) } : undefined,
|
||||
});
|
||||
}));
|
||||
getNativeFreshSessionHandoff = nativeReviewRequest ? undefined : getFreshSessionHandoff;
|
||||
const nativeProviderConfig = nativeRuntimeResolution.profile.backend === "cursor_cloud_api"
|
||||
|| nativeRuntimeResolution.profile.backend === "codex_app_server"
|
||||
@@ -761,12 +761,12 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele
|
||||
adapterConfig: nativeProviderConfig, managedProfile, agentCoreProfile, dotBinding, museBinding,
|
||||
});
|
||||
const codexCliVersion = agent.adapterType === "paperclip_runner" && requestedNativeProvider.provider === "codex"
|
||||
? await readRemoteCodexModelCliVersion({
|
||||
? await measureRunPhase("runtime_selection.codex_cli_version", () => readRemoteCodexModelCliVersion({
|
||||
model: requestedNativeProvider.model,
|
||||
target: executionTarget,
|
||||
remoteCodexPath: runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_PATH,
|
||||
remoteCodexNpmSpec: runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_NPM_SPEC,
|
||||
}) : null;
|
||||
})) : null;
|
||||
const nativeProvider = codexCliVersion
|
||||
? resolvePaperclipRunnerNativeProviderInput({
|
||||
companyId: agent.companyId,
|
||||
|
||||
Reference in new issue
Block a user