From d95fdd8d79e0db0e9e83386f1162cbf01ef44e2f Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Mon, 9 Nov 2026 15:27:58 -0800 Subject: [PATCH] feat(observability): span the unnamed gaps in run preparation (#20396) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Before a task reaches the agent, the server prepares the run: it reads the task, the workspace and the configuration, and then hands the run to the native controller > - With an OTLP endpoint, each preparation step emits a `heartbeat.prepare.` or `native.controller.` span > - On a hosted test deployment, 0.3 to 0.5 s of each `task.prepare` sits between those spans, and about 0.5 s more sits between `execution.dispatch_admission` and `native.coordinator.claim` > - Nobody can say which step is slow in that time, so latency work in this area is guesswork > - This pull request adds spans around the existing awaited work in those gaps, with the existing timing helpers and naming conventions > - The benefit is that a trace of a slow preparation now names the step that used the time ## Linked Issues or Issue Description **What existing behavior does this improve?** The OpenTelemetry trace of task preparation. Today it has unnamed gaps between the named preparation spans. **Subsystem affected** Server: heartbeat run preparation and the native controller (`server/src/services/heartbeat.ts`, `server/src/services/native-runtime/native-session-executor.ts`). **Current behavior** On a hosted test deployment (8 runs per group), `task.prepare` takes 2.6 to 3.4 s. Between `heartbeat.prepare.runtime_state` and `heartbeat.prepare.run_identity`, about 0.25 s has no span. Between `execution.dispatch_admission` and `native.coordinator.claim`, about 0.5 s has no span, and in one pair of runs it was 1.4 s. Nobody can tell which step used that time. **Proposed behavior** Spans name the existing awaited reads and writes in those gaps. The change adds no reads and records no content. The operator endpoint gate and the no-operation behaviour stay. **Reason and benefit** Latency work in this area needs a measurement. With these spans, a trace of a slow preparation names the step that used the time, so a fix can be confirmed or rejected after deployment. **Breaking changes** None. The change adds span names to the closed list. It changes no API, schema or behaviour. **Additional context** The server uses postgres.js, which the auto-instrumentation does not cover, so database spans do not fill these gaps. ## What Changed - `server/src/services/execution-preparation-timing.ts`: add the new closed phase names. - `server/src/services/heartbeat.ts`: add `heartbeat.prepare.*` spans for `chat_retry_admission`, `workspace_selection`, `dependency_readiness`, `issue_checkout`, `instance_settings`, `workspace_context`, `project_context`, `plan_wake_routing`, `routine_environment`, `responsible_user` and `execution_start`. The re-reads of the issue after a checkout use the existing `issue_context` name. - `server/src/services/native-runtime/native-session-executor.ts`: add `native.controller.*` spans for `attachment_staging`, `warm_restore`, `span_persistence` (persistence of the preparation spans to the run log) and `state_migration` before the coordinator claim. - `doc/observability.md`: list the new spans. - `server/src/__tests__/heartbeat-preparation-gap-spans.test.ts`: run a real heartbeat through native preparation with an in-memory span exporter. Check that each new span is emitted and that the new preparation spans are children of `task.execution_attempt`. ## Verification - `pnpm --filter @paperclipai/server typecheck` - `npx vitest run src/__tests__/heartbeat-preparation-gap-spans.test.ts` (fails without the source change, passes with it) - `npx vitest run src/services/heartbeat/preparation-timing-context.test.ts src/services/execution-preparation-timing.test.ts src/services/native-runtime/native-session-executor.test.ts src/__tests__/heartbeat-context-compaction-execution.test.ts` ## Risks - Low risk. The change only wraps existing awaited calls. Values and errors pass through the timing helper unchanged, and a checkout conflict still reaches the same `catch`. - Phases of 50 ms or more inside `heartbeat.prepare.*` are also written to the run log, as for the existing phases. This adds a few run-log rows on slow runs. - Without an OTLP endpoint, the helpers keep their current no-operation behaviour. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, test runs) and extended thinking. ## 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 - [x] All Paperclip CI gates are green - [x] 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 --- doc/observability.md | 10 +- .../heartbeat-preparation-gap-spans.test.ts | 138 ++++++++++++++++++ .../services/execution-preparation-timing.ts | 6 +- server/src/services/heartbeat.ts | 110 +++++++------- .../native-runtime/native-session-executor.ts | 43 +++--- 5 files changed, 234 insertions(+), 73 deletions(-) create mode 100644 server/src/__tests__/heartbeat-preparation-gap-spans.test.ts diff --git a/doc/observability.md b/doc/observability.md index c68ca2820e3..69c9d899cdf 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -236,6 +236,12 @@ Early context preparation also records `agent_read`, `chat_completion`, `conversation`, `runtime_state`, `issue_context`, `run_identity`, `trust_preset`, and `task_context` beneath `heartbeat.prepare`. These time the existing awaited operations before environment selection; they do not add reads or record content. +The work between those phases is named too: `chat_retry_admission`, +`workspace_selection`, `dependency_readiness`, `issue_checkout`, +`instance_settings`, `workspace_context`, `project_context`, +`plan_wake_routing`, `routine_environment` and `responsible_user`. +`execution_start` times the run and agent status writes and the "run started" +event before run output starts. A repeated read or write has its own span. Execution configuration also records separate phases for mentioned skills, GitHub identity selection, secret resolution, managed AI runtime preparation, @@ -297,7 +303,9 @@ The `paperclip.native-controller` tracer records backend construction, connector assignments, assigned MCP discovery, tool authority, companion selection and Runner binary resolution. It also records GitHub shutdown, coordinator release, instruction checkpoints and collection, -and warm-session release. These intervals can occur between the existing native +and warm-session release. Before the coordinator claim it also records +attachment staging, warm controller restore, persistence of the preparation +spans to the run log, and Runner state migration. These intervals can occur between the existing native preparation, execution and settlement spans. They are controller work, not model turn time, and carry only the existing run hash. Nested phases overlap their containing backend or lifecycle intervals; do not add both durations. diff --git a/server/src/__tests__/heartbeat-preparation-gap-spans.test.ts b/server/src/__tests__/heartbeat-preparation-gap-spans.test.ts new file mode 100644 index 00000000000..e9e63f45ee1 --- /dev/null +++ b/server/src/__tests__/heartbeat-preparation-gap-spans.test.ts @@ -0,0 +1,138 @@ +import { randomUUID } from "node:crypto"; +import { context, trace } from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; +import { eq } from "drizzle-orm"; +import { agents, authUsers, companyMemberships, createDb, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import type { NativeRunIdentity, NativeSession, NativeSessionBackend, PrpEvent } from "@paperclipai/paperclip-runner"; +import { describe, expect, it, vi } from "vitest"; +import { heartbeatService, type HeartbeatServiceOptions } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { closeIdleWarmNativeSessionsForRestart } from "../services/native-runtime/native-session-executor.js"; +import { nativeRestartFixture, waitForCondition } from "./helpers/native-restart-fixture.js"; +import { getRecoveryPostgresSupport } from "./helpers/recovery-postgres-fixture.js"; + +const support = await getRecoveryPostgresSupport({}); + +function completingBackend(db: ReturnType): NonNullable { + return execution => { + let identity: NativeRunIdentity; + let sourceInstanceId = ""; + let turn = Promise.withResolvers(); + const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; + const session: NativeSession = { + identity: () => identity, + capabilities: async () => capabilities, + attachRun: async input => { identity = input.identity; }, + startTurn: async () => { + const turnId = randomUUID(); + turn.resolve(turnId); + return { turnId }; + }, + async *events() { + const turnId = await turn.promise; + const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record): PrpEvent => ({ + schema: "paperclip.prp.event.v1", sourceKind: "runner", sourceInstanceId, + sourceEventId: `${sourceInstanceId}:${identity.runId}:${sourceSeq}`, sourceSeq, + runId: identity.runId, normalizedSessionId: identity.sessionId, turnId, + schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), eventType, payload, + }); + yield event(1, "turn.started", {}); + yield event(2, "item.completed", { kind: "agentMessage", channel: "final", text: "Done." }); + turn = Promise.withResolvers(); + yield event(3, "turn.completed", { status: "completed" }); + }, + result: async () => null, + snapshot: async () => ({ backendKind: "mock", sessionId: "gap-spans-session", identity, + providerSessionId: "gap-spans-provider-session", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }), + close: async () => {}, + }; + const backend: NativeSessionBackend = { + descriptor: async () => { + const [run] = await db.select({ runnerInstanceId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, execution.binding.runId)); + if (!run?.runnerInstanceId) throw new Error("Normal native admission did not persist a runner identity"); + sourceInstanceId = run.runnerInstanceId; + return { kind: "mock", name: "preparation-gap-spans", version: "1", capabilities, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }; + }, + openSession: async input => { identity = input.identity; return session; }, + recoverSession: async snapshot => { identity = snapshot.identity; return { recovered: true, session }; }, + }; + return backend; + }; +} + +describe.skipIf(!support.supported)("heartbeat preparation gap spans", () => { + const fixture = nativeRestartFixture({}); + + it("emits spans for the work between the named preparation phases", async () => { + const { companyId, agentId, temporary } = fixture.current; + const db = createDb(temporary.connectionString, { maxConnections: 4 }); + const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT; + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test"); + const manager = new AsyncLocalStorageContextManager().enable(); + const exporter = new InMemorySpanExporter(); + const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] }); + expect(context.setGlobalContextManager(manager)).toBe(true); + expect(trace.setGlobalTracerProvider(provider)).toBe(true); + const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: completingBackend(db) }); + try { + const userId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Task owner", email: `${userId}@example.invalid`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" }); + await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex", lifecycleMode: "warm", idleTimeoutMs: 60_000 } }) + .where(eq(agents.id, agentId)); + await instanceSettingsService(db).updateExperimental({ enableAgentChat: true, enableNativeRunner: true }); + const [issue] = await db.insert(issues).values({ companyId, title: "Trace the preparation gaps", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: userId, conversationAgentId: agentId, conversationUserId: userId, + conversationState: "active" }).returning(); + const [message] = await db.insert(issueComments).values({ companyId, issueId: issue!.id, + authorType: "user", authorUserId: userId, body: "Start." }).returning(); + const run = await heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: userId, manualUserWake: true, + payload: { issueId: issue!.id, commentId: message!.id }, + contextSnapshot: { issueId: issue!.id, taskKey: issue!.id, wakeCommentId: message!.id } }); + expect(run).not.toBeNull(); + await waitForCondition("the run to settle", () => heartbeat.getTaskDrainStatus().quiescent); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id)))[0]) + .toMatchObject({ status: "succeeded" }); + + await provider.forceFlush(); + const spans = exporter.getFinishedSpans(); + const names = new Set(spans.map(span => span.name)); + const preparation = [ + "heartbeat.prepare.chat_retry_admission", + "heartbeat.prepare.workspace_selection", + "heartbeat.prepare.dependency_readiness", + "heartbeat.prepare.instance_settings", + "heartbeat.prepare.workspace_context", + "heartbeat.prepare.plan_wake_routing", + "heartbeat.prepare.routine_environment", + "heartbeat.prepare.responsible_user", + "heartbeat.prepare.execution_start", + ]; + const controller = [ + "native.controller.attachment_staging", + "native.controller.warm_restore", + "native.controller.span_persistence", + "native.controller.state_migration", + ]; + for (const name of [...preparation, ...controller]) expect(names, name).toContain(name); + const attempts = spans.filter(span => span.name === "task.execution_attempt"); + expect(attempts).toHaveLength(1); + for (const span of spans.filter(span => preparation.includes(span.name))) { + expect(span.parentSpanContext?.spanId, span.name).toBe(attempts[0]!.spanContext().spanId); + } + } finally { + await heartbeat.drainActiveRunExecutions(); + await closeIdleWarmNativeSessionsForRestart(); + await db.$client.end(); + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint); + trace.disable(); + context.disable(); + await provider.shutdown(); + } + }, 60_000); +}); diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index d8753197a72..01041233d1a 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -7,6 +7,9 @@ import { isSpanContextValid, ROOT_CONTEXT, SpanStatusCode, trace, type Context } type PreparationPhase = "environment_selection" | "execution_configuration" | "session" | "agent_read" | "chat_completion" | "conversation" | "runtime_state" | "issue_context" | "run_identity" | "trust_preset" | "task_context" + | "chat_retry_admission" | "workspace_selection" | "dependency_readiness" | "issue_checkout" + | "instance_settings" | "workspace_context" | "project_context" | "plan_wake_routing" + | "routine_environment" | "responsible_user" | "execution_start" | "workspace" | "github_environment" | "github_launchers" | "run_output" | "workspace_validation" | "runtime_services" | "instructions" | "runtime_selection" | "configuration_mentions" | "configuration_github_identity" | "configuration_secrets" @@ -34,7 +37,8 @@ type PreparationPhase = "environment_selection" | "execution_configuration" | "s | "github_access" | "wake_binding" | "backend_setup" | "connector_assignments" | "assigned_mcp" | "tool_authority" | "companion_selection" | "runner_binary" | "github_stop" | "coordinator_release" | "instruction_checkpoint" - | "instruction_changes" | "warm_release" | "instruction_collection"; + | "instruction_changes" | "warm_release" | "instruction_collection" + | "attachment_staging" | "warm_restore" | "span_persistence" | "state_migration"; export type ExecutionPhase = | "skills.inventory" | "skills.inventory.wait" | "skills.inventory.bundled" | "skills.inventory.releases" diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e00e1c6f1c1..bb08881e332 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2153,7 +2153,7 @@ export function heartbeatService( contextSnapshot: context, })); }); - const isFailedChatRunRetry = await authorizeFailedChatRetryExecution(); + const isFailedChatRunRetry = await timePreparation("chat_retry_admission", () => authorizeFailedChatRetryExecution()); // Never adopt a chat-execution attestation supplied in a wake payload. // Reviewed chat turns rebuild it from the current durable owner below. delete context[PAPERCLIP_EXTERNAL_CHAT_EXECUTION_BOUND_KEY]; @@ -2208,28 +2208,31 @@ export function heartbeatService( const taskKey = deriveTaskKeyWithHeartbeatFallback(context, null); const sessionCodec = getAdapterSessionCodec(agent.adapterType); const issueId = readNonEmptyString(context.issueId); - const freshWorkspace = await readFreshWorkspaceReadiness(db, run); - if (issueId && canApplyTaskWorkspaceSelectionAtAdmission({ admittedInput: parseObject(run.runnerProfileJson).nativeExecutionInput, restarting: Boolean(runOptions.nativeRestartRecovery), hasLeaseOwner: Boolean(runOptions.nativeLeaseOwner) })) { - const changed = await executionWorkspacesSvc.applyPendingTaskWorkspaceSelection({ companyId: agent.companyId, - issueId, runId: run.id, actor: { type: "agent", agentId: agent.id, companyId: agent.companyId, source: "agent_key" } }); - // A subtask an older release bound to its parent's workspace gets its own - // here. A process-loss retry or a continuation that keeps the current - // workspace still finishes the interrupted attempt there first. - const released = run.processLossRetryCount > 0 || (freshWorkspace && freshWorkspace.mode !== "fresh_isolated_workspace") ? false - : await executionWorkspacesSvc.releaseAncestorWorkspaceBindingAtAdmission({ companyId: agent.companyId, issueId, runId: run.id }); - if (changed || released) { - delete context.resumeSessionParams; - delete context.resumeSessionDisplayId; - context.workspaceSelectionChanged = true; + const freshWorkspace = await timePreparation("workspace_selection", async () => { + const readiness = await readFreshWorkspaceReadiness(db, run); + if (issueId && canApplyTaskWorkspaceSelectionAtAdmission({ admittedInput: parseObject(run.runnerProfileJson).nativeExecutionInput, restarting: Boolean(runOptions.nativeRestartRecovery), hasLeaseOwner: Boolean(runOptions.nativeLeaseOwner) })) { + const changed = await executionWorkspacesSvc.applyPendingTaskWorkspaceSelection({ companyId: agent.companyId, + issueId, runId: run.id, actor: { type: "agent", agentId: agent.id, companyId: agent.companyId, source: "agent_key" } }); + // A subtask an older release bound to its parent's workspace gets its own + // here. A process-loss retry or a continuation that keeps the current + // workspace still finishes the interrupted attempt there first. + const released = run.processLossRetryCount > 0 || (readiness && readiness.mode !== "fresh_isolated_workspace") ? false + : await executionWorkspacesSvc.releaseAncestorWorkspaceBindingAtAdmission({ companyId: agent.companyId, issueId, runId: run.id }); + if (changed || released) { + delete context.resumeSessionParams; + delete context.resumeSessionDisplayId; + context.workspaceSelectionChanged = true; + } } - } + return readiness; + }); let issueContext = issueId ? await timePreparation("issue_context", () => getIssueExecutionContext(agent.companyId, issueId)) : null; const issueDependencyReadiness = issueId - ? await issuesSvc + ? await timePreparation("dependency_readiness", () => issuesSvc .listDependencyReadiness(agent.companyId, [issueId]) - .then((rows) => rows.get(issueId) ?? null) + .then((rows) => rows.get(issueId) ?? null)) : null; const dependencyResponse = issueId && issueContext?.status === "blocked" && context.dependencyBlockedInteraction === true @@ -2247,14 +2250,17 @@ export function heartbeatService( // dispatch: an operator parking the issue after claim but before this // checkout must not be overwritten by the continuation. A todo task // that has never started has nothing parked, so it may start here. - const startsUnstartedTask = issueContext.status === "todo" && - await isUnstartedTaskQuestionAnswer(db, { companyId: agent.companyId, issueId, interactionId: context.interactionId }); - await issuesSvc.checkout( - issueId, - agent.id, - [...resolvedInteractionCheckoutExpectedStatuses(), ...(startsUnstartedTask ? ["todo"] : [])], - run.id, - ); + const issueStatus = issueContext.status; + await timePreparation("issue_checkout", async () => { + const startsUnstartedTask = issueStatus === "todo" && + await isUnstartedTaskQuestionAnswer(db, { companyId: agent.companyId, issueId, interactionId: context.interactionId }); + await issuesSvc.checkout( + issueId, + agent.id, + [...resolvedInteractionCheckoutExpectedStatuses(), ...(startsUnstartedTask ? ["todo"] : [])], + run.id, + ); + }); context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true; } catch (error) { if (!isCheckoutConflictError(error)) throw error; @@ -2269,7 +2275,7 @@ export function heartbeatService( } throw error; } - issueContext = await getIssueExecutionContext(agent.companyId, issueId); + issueContext = await timePreparation("issue_context", () => getIssueExecutionContext(agent.companyId, issueId)); } if ( issueId && @@ -2286,18 +2292,18 @@ export function heartbeatService( }) ) { try { - await issuesSvc.checkout( + await timePreparation("issue_checkout", () => issuesSvc.checkout( issueId, agent.id, typeof context.workspaceReconstructionActionId === "string" ? ["todo"] : ["todo", "backlog", "blocked"], run.id, - ); + )); context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true; } catch (error) { if (!isCheckoutConflictError(error)) throw error; context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = false; } - issueContext = await getIssueExecutionContext(agent.companyId, issueId); + issueContext = await timePreparation("issue_context", () => getIssueExecutionContext(agent.companyId, issueId)); } if ( issueId && @@ -2338,7 +2344,7 @@ export function heartbeatService( ) : null; const experimentalInstanceSettings = - await instanceSettings.getExperimental(); + await timePreparation("instance_settings", () => instanceSettings.getExperimental()); const isolatedWorkspacesEnabled = experimentalInstanceSettings.enableIsolatedWorkspaces; // Inert on its own: the operator default only reaches the resolver when @@ -2349,16 +2355,18 @@ export function heartbeatService( experimentalInstanceSettings.enableIsolatedWorkspacesByDefault; const providerApiPlacement = resolveProviderApiPlacement({ run, agent }); const externalProvider = heartbeatExternalProvider({ run, agent }) ?? providerApiPlacement?.provider; - const [taskWorkspaceIntentRow] = issueId && !externalProvider ? await db.select({ intent: issues.workspaceSelection }).from(issues) - .where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId))) : []; + const [taskWorkspaceIntentRow] = issueId && !externalProvider ? await timePreparation("workspace_context", () => db.select({ intent: issues.workspaceSelection }).from(issues) + .where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)))) : []; const contextProjectId = readNonEmptyString(context.projectId); - const boundSourceWorkspace = !externalProvider && issueContext?.executionWorkspaceId - ? await executionWorkspacesSvc.getStoredById(agent.companyId, issueContext.executionWorkspaceId) : null; - const [selectedWorkspaceSource] = !externalProvider && !boundSourceWorkspace && issueContext?.projectWorkspaceId - ? await db.select({ projectId: projectWorkspaces.projectId }).from(projectWorkspaces) - .where(and(eq(projectWorkspaces.id, issueContext.projectWorkspaceId), eq(projectWorkspaces.companyId, agent.companyId))) : []; + const executionWorkspaceId = issueContext?.executionWorkspaceId; + const boundSourceWorkspace = !externalProvider && executionWorkspaceId + ? await timePreparation("workspace_context", () => executionWorkspacesSvc.getStoredById(agent.companyId, executionWorkspaceId)) : null; + const projectWorkspaceId = issueContext?.projectWorkspaceId; + const [selectedWorkspaceSource] = !externalProvider && !boundSourceWorkspace && projectWorkspaceId + ? await timePreparation("workspace_context", () => db.select({ projectId: projectWorkspaces.projectId }).from(projectWorkspaces) + .where(and(eq(projectWorkspaces.id, projectWorkspaceId), eq(projectWorkspaces.companyId, agent.companyId)))) : []; const loadProjectContext = async (projectId: string | null | undefined) => projectId - ? await db + ? await timePreparation("project_context", () => db .select({ id: projects.id, executionWorkspacePolicy: projects.executionWorkspacePolicy, @@ -2385,7 +2393,7 @@ export function heartbeatService( eq(projects.companyId, agent.companyId), ), ) - .then((rows) => rows[0] ?? null) + .then((rows) => rows[0] ?? null)) : null; const taskProjectId = issueContext ? issueContext.projectId : contextProjectId; const taskProjectContext = await loadProjectContext(taskProjectId); @@ -2407,14 +2415,14 @@ export function heartbeatService( readNonEmptyString(context.interactionStatus) === "accepted") : false; const acceptedPlanWakeRoutingDecision = issueContext - ? await resolveAcceptedPlanWakeRoutingDecision({ + ? await timePreparation("plan_wake_routing", () => resolveAcceptedPlanWakeRoutingDecision({ db, companyId: agent.companyId, agentId: agent.id, issueId, acceptedPlanContinuationWake, contextSnapshot: context, - }) + })) : null; if (acceptedPlanWakeRoutingDecision) { context.forceFreshSession = true; @@ -2434,17 +2442,17 @@ export function heartbeatService( } else { delete context.acceptedPlanWakeRouting; } - const routineEnvContext = await getRoutineEnvForExecutionIssue( + const routineEnvContext = await timePreparation("routine_environment", () => getRoutineEnvForExecutionIssue( agent.companyId, issueContext, - ); + )); let responsibleUserId: string | null = - await resolveResponsibleUserIdForRun({ + await timePreparation("responsible_user", () => resolveResponsibleUserIdForRun({ run, contextSnapshot: context, issueContext, routineEnvContext, - }); + })); const identityContext = await timePreparation("run_identity", () => initializeRunIdentity(db, { companyId: agent.companyId, runId: run.id, @@ -3231,7 +3239,7 @@ export function heartbeatService( } = { current: null }; try { const startedAt = run.startedAt ?? new Date(); - const runningWithSession = await db + const runningWithSession = await timePreparation("execution_start", () => db .update(heartbeatRuns) .set({ startedAt, @@ -3242,12 +3250,12 @@ export function heartbeatService( }) .where(eq(heartbeatRuns.id, run.id)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (runningWithSession) run = runningWithSession; // Pause Durability: flip to "running" ONLY if the agent is still invokable. // Atomic conditional UPDATE is the sole gate (no read-then-write); 0 rows => abort. - const runningAgent = await db + const runningAgent = await timePreparation("execution_start", () => db .update(agents) .set({ status: "running", updatedAt: new Date() }) .where( @@ -3258,7 +3266,7 @@ export function heartbeatService( ), ) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (!runningAgent) { logger.warn( @@ -3300,12 +3308,12 @@ export function heartbeatService( }); const currentRun = run; - await appendRunEvent(currentRun, { + await timePreparation("execution_start", () => appendRunEvent(currentRun, { eventType: "lifecycle", stream: "system", level: "info", message: "run started", - }); + })); const { currentUserRedactionOptions, appendIdentityRedactedLog, onLog, onPersistedNativeLog, onAdapterMeta, onAdapterEvent } = await timePreparation("run_output", () => runOutput.begin(currentRun)); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e8df4b0b52f..f0e9a0e335f 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9533,9 +9533,10 @@ export async function executePaperclipNativeSession(input: { } return preparedInput; }; + const stageAttachments = () => createRunPreparationTiming(runId, "native.controller")("attachment_staging", prepareAttachments); return needsWarmRetirement - ? await executePaperclipNativeSessionWithinScope(input, prepareAttachments, false, ownDiagnosticLease) - : await executePaperclipNativeSessionWithinScope(await prepareAttachments(), undefined, false, ownDiagnosticLease); + ? await executePaperclipNativeSessionWithinScope(input, stageAttachments, false, ownDiagnosticLease) + : await executePaperclipNativeSessionWithinScope(await stageAttachments(), undefined, false, ownDiagnosticLease); } catch (error) { executionFailure = error; if (error instanceof NativeControllerRecoveryWaitingError) { @@ -9692,7 +9693,8 @@ async function executePaperclipNativeSessionAttempt( : null; const stagedRestart = input.restartRecovery?.kind === "reattach_remote_runner" ? await verifyStagedAttachmentRestart(input) : null; - if (input.useRunnerd) await restoreWarmRunnerdController(input); + const timeController = createRunPreparationTiming(input.execution.binding.runId, "native.controller"); + if (input.useRunnerd) await timeController("warm_restore", () => restoreWarmRunnerdController(input)); const identityRedactor = createAgentIdentityRedactor(input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY); const redactIdentityText = (text: string) => redactSensitiveText( identityRedactor.chunk("diagnostic", text) + identityRedactor.finish("diagnostic"), @@ -9718,7 +9720,6 @@ async function executePaperclipNativeSessionAttempt( "paperclip_runner_provider_unsupported: ACPX candidate requires exact host qualification authorization", ); } - const timeController = createRunPreparationTiming(input.execution.binding.runId, "native.controller"); const preparationSpans = input.preparationSpans ?? []; const preparationStarts = nativeRunPreparationStarts( preparationSpans, @@ -9772,20 +9773,22 @@ async function executePaperclipNativeSessionAttempt( startedAtMs: environmentStartedAtMs, }) : null; - for (const span of preparationSpans) { - const rootMilestone = isNativeRunRootHistoricalSpan(span.name); - await trace.record({ - ...span, - parentName: rootMilestone - ? "task.provider_session" - : environmentSpans.includes(span) - ? "environment.startup" - : "task.prepare", - }); - } - if (environmentScope) { - await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs }); - } + await timeController("span_persistence", async () => { + for (const span of preparationSpans) { + const rootMilestone = isNativeRunRootHistoricalSpan(span.name); + await trace.record({ + ...span, + parentName: rootMilestone + ? "task.provider_session" + : environmentSpans.includes(span) + ? "environment.startup" + : "task.prepare", + }); + } + if (environmentScope) { + await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs }); + } + }); let retainedTransition: VerifiedWarmTransitionBinding | undefined; let remoteWarmRecovery: RemoteWarmSessionRecovery | undefined; const cachedReleasedOwner = warmNativeSessions.get(nativeSessionScopeKey(input.execution)); @@ -9870,7 +9873,7 @@ async function executePaperclipNativeSessionAttempt( } } if (input.useRunnerd) { - retainedTransition = await migrateRunnerdStateRootForExecution({ + retainedTransition = await timeController("state_migration", () => migrateRunnerdStateRootForExecution({ db: input.db, execution: input.execution, preparedExecutionAdmission: input.preparedExecutionAdmission, @@ -9881,7 +9884,7 @@ async function executePaperclipNativeSessionAttempt( restartRecovery: input.restartRecovery, runnerExecutionTarget: input.runnerExecutionTarget, bootstrapRetryLeaseOwner: input.leaseOwner, - }); + })); } const durableRunnerBinding = input.useRunnerd ? loadRunnerdDurableBinding(input.execution, retainedTransition)