From 25b16e65fc70a617a70e3c5f50ce358e333615d8 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 15 Oct 2026 12:21:32 -0500 Subject: [PATCH] fix(observability): detail workspace and connector preparation (#16454) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The server prepares a repository, sandbox workspace, and connector access before an agent can reply. > - Existing traces measure large preparation phases but leave important work inside those phases unclear. > - Repository source materialization and Browser Use discovery can do more work than their callers suggest. > - New detail must preserve access checks, run ownership, and the existing work order. > - This pull request adds fixed OpenTelemetry spans inside those operations. > - Operators can use one deployed trace to select the next performance fix. ## Linked Issues or Issue Description Refs: #16419, #16425, #16428. **What existing behavior does this improve?** Startup traces for server workspace preparation, native sandbox synchronization, and connector assignment resolution. **Current behavior** Workspace and connector aggregate spans can take many seconds. The trace does not separate private source cloning, seed reuse, archive work, progress persistence, or each connector resolver. Browser Use resource selection builds the full authorized tool catalog. The configuration and native backend can each resolve connector assignments. **Proposed behavior** Show the awaited work as fixed child spans. Separate discovery queue wait from catalog construction. Preserve the initiating run context when another run releases a scheduler slot. Add no new lifecycle event writes for workspace detail callbacks. **Reason and benefit** Operators need evidence about the hosted critical path before another performance change. These spans distinguish filesystem work, provider transfer, authorization, transaction entry, locks, and reporting overhead. **Breaking changes** None. Authorization, query order, transaction ownership, synchronization, existing progress events, and task results retain their contracts. The change adds opt-in Observability spans under the existing OTLP endpoint gate. ## What Changed - Trace private source directory setup, Git credentials, clone, requested ref, source identity, and checkout publication. - Trace sandbox source capture, seed cache/replay, shallow clone and archive, overlay copy/archive, verification, upload, warm stamps, and descriptor persistence. - Separate existing awaited progress reporting from useful workspace work. Keep new detail callbacks trace-only. - Trace each connector resolver and Browser Use catalog/grant work. Split discovery queueing, descriptor sources, and policy evaluation. - Trace Slack authority, credential reads, catalog synchronization, transaction entry/settlement, and advisory locks. - Update the real filesystem staging trace expectations and the Honeycomb investigation guide. - Add a real scheduler regression for a queued job released by another run. Assert run ownership, parent spans, and exactly one completed queue span. ## Verification - Passed: server and adapter-utils TypeScript checks (`pnpm --filter @paperclipai/server exec tsc --noEmit` and `pnpm --filter @paperclipai/adapter-utils exec tsc --noEmit`). - Passed: 252 checks across 9 focused files. These cover real filesystem staging, native workspace synchronization, trace ownership, endpoint gating, and workspace preparation. - Passed after review: 26 trace-context, tracing-failure, and discovery-scheduler checks across 3 files (these overlap the first run). The new queued-discovery regression fails when context binding is removed and passes after restoration. An initial submillisecond timestamp assertion was replaced with explicit span-completion order to avoid SDK clock precision assumptions. - Passed on implementation commit `0e68968a4bd61001893d833f281f12bb110edbe1`: `pnpm -r typecheck` and `pnpm build`. The later commit updates only an engine test expectation. - The first CI workspace lane found an old engine trace-tree expectation. It expected the earlier span set. The updated assertion now checks the new detail spans, their parents, and single completion. Its targeted executor check passes. - The full test suite, build, release registry, and end-to-end gates pass in CI on final head `40e37d89ef679109d60a96bb5731a0616efeaa43`: 55 passing checks and 2 optional Storybook skips. The full suite was not duplicated locally after the focused run. - Greptile reviewed the final head at 5/5. No unresolved review threads remain. - After deployment, filter Honeycomb by the hashed run ID and verify `service.version`. Inspect `heartbeat.prepare.workspace`, `runtime_selection.workspace_sync`, and the connector assignment parents. The guide in `doc/observability.md` explains each new family and the nested intervals. - No hosted run has used this new instrumentation yet. Local checks cannot establish hosted latency improvement. ## Risks - Tracing adds spans when an operator configures an OTLP endpoint. Detail remains capped at 16 spans per phase per execution attempt. - Parent and child durations overlap. The documentation explains how to avoid counting them twice. - A discovery job can start in another run's scheduler continuation. The bound callback preserves the initiating async context. - New spans export fixed names and the existing hashed run ID. They add no paths, repository URLs, connector resource IDs, credentials, tool schemas, or exception text. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, tool use, and code execution. The exact served model ID and context-window size are not exposed in this session. ## 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 --------- Co-authored-by: Paperclip --- doc/observability.md | 48 ++++++++++- .../src/acpx-engine/execute.test.ts | 43 ++++++---- .../src/sandbox-managed-runtime.test.ts | 11 ++- .../src/sandbox-managed-runtime.ts | 78 ++++++++--------- .../chat-channels/slack/task-authority.ts | 12 +-- server/src/services/connector-runtime.ts | 11 ++- .../src/services/connectors/slack-catalog.ts | 17 ++-- server/src/services/connectors/slack.ts | 16 ++-- .../services/execution-preparation-timing.ts | 83 +++++++++++++++++-- .../services/execution-timing-context.test.ts | 51 +++++++++++- .../services/execution-transaction-timing.ts | 2 +- server/src/services/heartbeat.ts | 7 +- .../heartbeat/workspace-preparation.ts | 18 ++-- .../isolated-remote-workspace-source.ts | 46 +++++----- .../services/managed-repository-checkout.ts | 18 ++-- .../native-runtime/native-workspace-sync.ts | 41 +++++---- server/src/services/tool-gateway.ts | 33 ++++---- 17 files changed, 371 insertions(+), 164 deletions(-) diff --git a/doc/observability.md b/doc/observability.md index 2fcf0ade1b8..48f417cc2dc 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -254,10 +254,22 @@ attempt or owning phase: | `workspace.sources.*` | Source inventory queries, canonicalization, one filesystem alias scan, repository scan, and configured project scan. | | `workspace.readers`, `workspace.readers.*`, `workspace.access.*` | Retained-reader groups, identity, trust, project and source authorization, privacy and inventory reads. | | `workspace.provenance`, `workspace.notification`, `workspace.write` | Preserved ownership, notification statements, and record writes. | +| `workspace.materialize.*`, `workspace.checkout.*` | Private source-directory setup and identity verification, existing checkout lookup, Git credential resolution, clone, requested-ref checkout, and atomic publication. The enclosing `heartbeat.prepare.workspace_source_materialize` includes the complete source-materialization path. | | `runtime_selection.*` | Completion contracts, session/recovery reads, profiles, interaction materialization, backend discovery, bootstrap, run/coordinator writes, transaction entry/settlement, and workspace synchronization. | | `runtime_context.instructions`, `.skills`, `.mcp` | The three concurrent runtime-context branches. Their durations overlap. | | `runtime_context.asset_*` | Source reads, manifest hashing, cache lookup, cold materialization, and exact file verification on cold and warm paths. | -| `workspace_sync.*` | Existing Git and baseline snapshot, pack, seed verification, transfer, and restore operations. Pack and verification can be inside transfer; do not add nested durations. | +| `workspace_sync.prepare_host_current`, `.prepare_durable_seed`, `.prepare_adopt_remote` | Runtime preparation under the selected inbound mode, including snapshot and staging work. A warm mismatch may produce a second preparation; it is real repeated work. | +| `workspace_sync.source_root`, `.repositories`, `.ignore_rules`, `.snapshot_git`, `.snapshot_baseline` | Physical source capture, managed repositories, ignore rules, Git metadata and baseline content snapshot. | +| `workspace_sync.pack`, `.git_clone`, `.git_archive`, `.overlay_copy`, `.overlay_archive`, `.deleted_manifest` | Temporary shallow Git clone, its archive callback, selected working-tree overlay copy/archive, and the deleted-file manifest. The archive is inside the clone's lifetime; do not sum parent and child. | +| `workspace_sync.seed_generation`, `.seed_lookup`, `.seed_replay`, `.seed_persist`, `.seed_publish`, `.seed_verify`, `.seed_extract`, `.seed_snapshot`, `.seed_cleanup` | Seed identity, cache lookup, replay, durable copies, publication, extraction/re-snapshot verification and temporary cleanup. Cache hits omit cold clone and verification work. | +| `workspace_sync.inbound_barrier`, `.stage_workspace`, `.transfer` | The inbound scheduler barrier, whole workspace task, and provider `syncIn` call (including provider guards, upload and post-upload extraction). Parallel tasks overlap; provider transfer/exec children give the finer transport split when available. | +| `workspace_sync.progress` | Awaited progress/status reporting, including the existing lifecycle run-log writes. These are not file transfer work. | +| `workspace_sync.run_read`, `.descriptor_directory`, `.descriptor_read`, `.archive_hash`, `.warm_stamp`, `.durable_seed_verify`, `.manifests_persist`, `.descriptor_write`, `.reference_write` | Saved run state, recovery descriptors, archive hashing, remote warm-stamp check, manifests and final descriptor/reference persistence. | +| `connectors.browser_use.resolve`, `connectors.speko.resolve`, `connectors.slack.resolve`, `connectors.agentmail.resolve` | Each sequential connector resource resolver beneath configuration or native-controller assignment resolution. Empty/denied assignments still have measurable work. | +| `connectors.browser_use.connection`, `.catalog`, `.grant` | Browser Use candidate reads, the full authorized catalog listing currently used to discover Browser Use resources, and grant checks. The catalog can contain work for other connectors. | +| `connectors.discovery.scheduled`, `.queue`, `.build`, `.agent`, `.connections`, `.github`, `.slack`, `.speko`, `.policy`, `.on_demand_policy` | Discovery scheduler lifetime and wait, catalog construction, agent validation, descriptor sources, and policy decisions. Queued work retains the initiating run's trace context even when another run releases the scheduler slot. | +| `connectors.slack.candidates`, `.authority`, `.profile`, `.settings`, `.identity`, `.credentials`, `.revision` | Slack endpoint enumeration, authority revalidation, enabled profile, settings, accepted requester provenance, credential reads and authorization revision. | +| `connectors.slack.authority.transaction*`, `connectors.slack.catalog.transaction*`, `connectors.slack.catalog.lock`, `.entries`, `.write`, `.profile` | Slack authority and catalog transaction entry/settlement, advisory lock, catalog reads/updates and existing-profile lookup. Entry combines pool acquisition and BEGIN; it does not isolate pool wait. | | `execution.runtime` | Runtime dispatch, usage, workspace finalization and runtime cleanup; this includes work after provider execution. | | `execution.workspace_finalize` | Best-effort native workspace finalization, including restore recovery, copyback and commit. Its nested restore and commit spans measure distinct steps. | | `execution.workspace_ownership`, `.workspace_ownership_wait`, `.workspace_restore`, `.workspace_commit`, `.native_finalize` | Ownership coordination including its action, acquisition before the action, copyback, final snapshot/stamp/descriptor persistence, and native settlement. | @@ -283,6 +295,40 @@ can continue through later bookkeeping and resource cleanup. A later recovery invocation has its own attempt; this does not manufacture one continuous trace across queueing, process restarts, or an unavailable controller. +### Reading a slow startup in Honeycomb + +Filter `paperclip.task.run.run_id` to the first 12 hex characters of +`SHA256(run UUID)`, then inspect the attempt's waterfall with its +`service.version`. An agent/model setting or the repository's checked-out commit +does not establish the server revision. The detailed spans are available only +after this instrumentation is deployed; an earlier aggregate cannot establish +which child was slow. + +Start beneath `heartbeat.prepare.workspace` with source materialization versus +provisioning, repository admission, and environment acquisition/realization. +Source materialization now separates filesystem setup, credential resolution, +Git cloning and source identity checks. Under +`runtime_selection.workspace_sync`, compare snapshot work, the seed-cache path, +archive construction/verification, provider transfer and descriptor persistence. +Use span start/end times to identify serial work and uncovered gaps; summing +nested or concurrent durations overstates elapsed time. + +For connector delay, compare the four `connectors.*.resolve` spans beneath each +assignment-resolution parent. A long Browser Use `.catalog` can be discovery +queueing, connection descriptors, another connector's authority checks, or policy +evaluation. A long Slack transaction entry, lock or credential span calls for a +different investigation. Configuration and native backend setup currently resolve +assignments separately; the trace records both without caching authorization or +changing access decisions. + +New workspace detail callbacks have fixed names and remain trace-only. They do +not create additional lifecycle events or awaited run-log writes. Existing coarse +progress writes remain intact and appear under `workspace_sync.progress`, so +reporting overhead is visible alongside useful work. No per-file spans, paths, +repository URLs, connector resource IDs, tool schemas, credentials or exception +messages are added to the export. The endpoint gate and per-phase span limit +above still apply. + ## Native Runner Trace Spans diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 422ad5a49e2..fe6021cc385 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -5275,7 +5275,7 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () = }; } - it("emits one root span with a child span per bring-up boundary, each parented to the root", async () => { + it("emits one root with startup and nested workspace boundaries", async () => { const root = await makeTempRoot(); const stateDir = path.join(root, "state"); const localCwd = path.join(root, "worktree"); @@ -5334,12 +5334,6 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () = expect(syncBackSpan!.parent).toBe(runRootSpan); expect(syncBackSpan!.ended).toBe(true); - // A codex bring-up over the remote sandbox lane crosses all 7 boundaries. - // Each boundary span parents to the sandbox bring-up span, not to the run - // root or the turn span. The `stage.sync` step also opens the two pre-task - // sub-step spans — `snapshot.git` and `snapshot.baseline` — plus the - // `stage.workspace` inbound task span, and the `pack` span nests one level - // deeper under `stage.workspace`. const childNames = spans .filter( (span) => @@ -5362,16 +5356,21 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () = "snapshot.git", "stage.sync", "stage.workspace", + "workspace.ignore_rules", + "workspace.inbound_barrier", + "workspace.overlay_archive", + "workspace.progress", + "workspace.progress", + "workspace.repositories", "workspace.resolve", + "workspace.source_root", + "workspace.transfer", ], ); - // The two pre-task sub-step spans and the `stage.workspace` task span nest - // under the `stage.sync` step span (that host work runs inside the step), not - // directly under the bring-up span. The `pack` span nests under - // `stage.workspace`, because the host builds the tarball inside the task. const stageSyncSpan = spans.find((span) => span.name === "stage.sync"); const stageWorkspaceSpan = spans.find((span) => span.name === "stage.workspace"); + const inboundSpan = spans.find((span) => span.name === "workspace.inbound_barrier"); const packSpan = spans.find((span) => span.name === "pack"); const snapshotGitSpan = spans.find((span) => span.name === "snapshot.git"); const snapshotBaselineSpan = spans.find((span) => span.name === "snapshot.baseline"); @@ -5380,15 +5379,12 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () = expect(packSpan).toBeTruthy(); expect(snapshotGitSpan).toBeTruthy(); expect(snapshotBaselineSpan).toBeTruthy(); - expect(stageWorkspaceSpan!.parent).toBe(stageSyncSpan); + expect(stageWorkspaceSpan!.parent).toBe(inboundSpan); expect(packSpan!.parent).toBe(stageWorkspaceSpan); expect(snapshotGitSpan!.parent).toBe(stageSyncSpan); expect(snapshotBaselineSpan!.parent).toBe(stageSyncSpan); expect(packSpan!.ended).toBe(true); - // Every boundary step span parents to the sandbox bring-up span and ends. - // The `stage.sync` sub-step spans and the run-parented `sandbox.syncBack` - // span are the exceptions: they parent above. const nonStartupChildren = new Set([ packSpan, stageWorkspaceSpan, @@ -5396,6 +5392,23 @@ describe("ACPX engine sandbox-start spans (opt-in root + child parenting)", () = snapshotBaselineSpan, syncBackSpan, ]); + for (const [name, parent] of [ + ["workspace.source_root", stageSyncSpan], ["workspace.repositories", stageSyncSpan], + ["workspace.ignore_rules", stageSyncSpan], ["workspace.inbound_barrier", stageSyncSpan], + ["workspace.overlay_archive", packSpan], ["workspace.transfer", stageWorkspaceSpan], + ] as const) { + const span = spans.find(span => span.name === name); + expect(span?.parent, name).toBe(parent); + expect(span?.ended, name).toBe(true); + expect(span?.endCalls, name).toBe(1); + nonStartupChildren.add(span); + } + const progress = spans.filter(span => span.name === "workspace.progress"); + expect(progress.map(span => span.parent)).toEqual([packSpan, stageWorkspaceSpan]); + for (const span of progress) { + expect(span.endCalls).toBe(1); + nonStartupChildren.add(span); + } for (const span of spans) { if (span === runRootSpan || span === startupSpan || span === turnSpan) continue; if (nonStartupChildren.has(span)) continue; diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts index b2135d1b279..854f76e9102 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts @@ -2943,7 +2943,11 @@ describe("sandbox managed runtime", () => { // The workspace stage task opens its own `stage.workspace` span, and the // host tarball build opens the `pack` span inside it. The two pre-task // sub-steps stay ahead of the task. - expect(openedSpans).toEqual(["snapshot.git", "snapshot.baseline", "stage.workspace", "pack"]); + expect(openedSpans).toEqual([ + "workspace.source_root", "snapshot.git", "workspace.repositories", "workspace.ignore_rules", + "snapshot.baseline", "workspace.inbound_barrier", "stage.workspace", "pack", + "workspace.progress", "workspace.overlay_archive", "workspace.transfer", "workspace.progress", + ]); // The tarball build still lands the workspace inside the span, so the wrap // changes no staging behavior. await expect(readFile(path.join(remoteWorkspaceDir, "README.md"), "utf8")).resolves.toBe("workspace body\n"); @@ -3002,8 +3006,11 @@ describe("sandbox managed runtime", () => { expect(packSpan!.ended).toBe(true); // The workspace stage task opens its own `stage.workspace` span under // `stage.sync`, and the `pack` span nests under `stage.workspace`. - expect(workspaceSpan!.parentName).toBe("stage.sync"); + expect(workspaceSpan!.parentName).toBe("workspace.inbound_barrier"); expect(packSpan!.parentName).toBe("stage.workspace"); + expect(spans.find(span => span.name === "workspace.inbound_barrier")?.parentName).toBe("stage.sync"); + expect(spans.find(span => span.name === "workspace.overlay_archive")?.parentName).toBe("pack"); + expect(spans.find(span => span.name === "workspace.transfer")?.parentName).toBe("stage.workspace"); // The two pre-`pack` staging sub-steps nest under `stage.sync` the same way, // so the previously hidden gap at the head of the step is now attributed. diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.ts b/packages/adapter-utils/src/sandbox-managed-runtime.ts index 3e591d7dffe..7458f3953e4 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.ts @@ -1265,11 +1265,13 @@ export async function prepareSandboxManagedRuntime(input: { // span, because the teardown runs the restore inside that span. const runStepSpan = (name: string, work: () => Promise): Promise => input.runtimeSpan ? input.runtimeSpan(name, work) : work(); + const reportWorkspaceStatus = (phase: RuntimeStatusPhase, message: string) => + runStepSpan("workspace.progress", () => emitRuntimeStatus(input.onRuntimeProgress, phase, message)); // Resolve an existing workspace alias once, before reading its snapshot. // All subsequent work uses that root, so retargeting the alias cannot select // another repository. Staging also verifies the captured directory identity. - const workspaceRoot = syncWorkspace ? await captureWorkspaceSourceRoot(input.workspaceLocalDir) : null; + const workspaceRoot = syncWorkspace ? await runStepSpan("workspace.source_root", () => captureWorkspaceSourceRoot(input.workspaceLocalDir)) : null; if (workspaceRoot) input = { ...input, workspaceLocalDir: workspaceRoot.sourceDir }; // The git enumeration (`git status --ignored`, the HEAD diffs, `ls-files`). @@ -1282,7 +1284,7 @@ export async function prepareSandboxManagedRuntime(input: { readGitWorkspaceSnapshot(input.workspaceLocalDir), ) : null; - const repositories = gitSnapshot ? (gitSnapshot.repositories ?? []) : input.workspaceRepositories ?? (syncWorkspace ? await readManagedWorkspaceRepositories(input.workspaceLocalDir) : []); + const repositories = gitSnapshot ? (gitSnapshot.repositories ?? []) : input.workspaceRepositories ?? (syncWorkspace ? await runStepSpan("workspace.repositories", () => readManagedWorkspaceRepositories(input.workspaceLocalDir)) : []); const repositoryGitPaths = (children: typeof repositories, prefix = ""): string[] => children.flatMap(repo => { const relative = prefix ? `${prefix}/${repo.path}` : repo.path; return [`${relative}/.git`, ...repositoryGitPaths(repo.snapshot.repositories ?? [], relative)]; @@ -1315,7 +1317,7 @@ export async function prepareSandboxManagedRuntime(input: { // repository's ignore rules still govern which files may leave the host. // Use the same bounded, path-relative resolver as referenced project trees. const directoryIgnore = syncWorkspace && !gitSnapshot && input.workspaceFileMode !== "all" - ? await resolveReferencedSourceIgnore(input.workspaceLocalDir) + ? await runStepSpan("workspace.ignore_rules", () => resolveReferencedSourceIgnore(input.workspaceLocalDir)) : null; if (directoryIgnore?.kind === "failed") { throw new Error(`Workspace ignore scan failed: ${directoryIgnore.reason}`); @@ -1368,9 +1370,9 @@ export async function prepareSandboxManagedRuntime(input: { } const seedGeneration = input.workspaceSeedCacheDirectory && baselineSnapshot && workspaceInboundMode !== "durable_seed" - ? workspaceSeedGeneration(baselineSnapshot, gitSnapshot, repositories) : null; + ? await runStepSpan("workspace.seed_generation", async () => workspaceSeedGeneration(baselineSnapshot!, gitSnapshot, repositories)) : null; const cachedSeed = seedGeneration - ? await readWorkspaceSeedGeneration(input.workspaceSeedCacheDirectory!, seedGeneration) : null; + ? await runStepSpan("workspace.seed_lookup", () => readWorkspaceSeedGeneration(input.workspaceSeedCacheDirectory!, seedGeneration)) : null; const replaySeed = workspaceInboundMode === "durable_seed" ? input.workspaceDurableSeed : cachedSeed; // Every inbound staging step delegates to the provider through `client.syncIn`: @@ -1467,7 +1469,7 @@ export async function prepareSandboxManagedRuntime(input: { params.progressLabel, { sink: input.onRuntimeProgress, phase: params.statusPhase }, ); - const syncResult = await syncIn(operations); + const syncResult = await runStepSpan("workspace.transfer", () => syncIn(operations)); // Prefer the transport's own byte total. It is the real count for a // provider that has no host tarball to stat (a referenced project rides // a `directory` mapping). Fall back to the caller-supplied count when @@ -1475,7 +1477,7 @@ export async function prepareSandboxManagedRuntime(input: { // shows the host-known workspace total. const transferredBytes = sumSyncResultBytes(syncResult); const reportedBytes = transferredBytes > 0 ? transferredBytes : params.progressBytes; - await upload.finish(reportedBytes, reportedBytes); + await runStepSpan("workspace.progress", () => upload.finish(reportedBytes, reportedBytes)); }; // Build the ordered inbound operation task list. Each task stages one inbound @@ -1522,7 +1524,7 @@ export async function prepareSandboxManagedRuntime(input: { // wipes the target tree EXCEPT `.paperclip-runtime`, so the overlay tar, // which sits under `.paperclip-runtime`, survives to run its own extract. if (gitSnapshot || repositories.length) { - await emitRuntimeStatus(input.onRuntimeProgress, "git_sync", "Syncing git history to environment"); + await reportWorkspaceStatus("git_sync", "Syncing git history to environment"); const gitTarPath = path.join(tempDir, "git-workspace.tar"); const remoteGitTar = path.posix.join( runtimeRootDir, @@ -1533,26 +1535,26 @@ export async function prepareSandboxManagedRuntime(input: { if (!durableGitArchive) { throw new Error("workspace_durable_seed_git_missing"); } - await copyDurableSeedArchive({ + await runStepSpan("workspace.seed_replay", () => copyDurableSeedArchive({ sourcePath: durableGitArchive, targetPath: gitTarPath, expectedSha256: replaySeed.gitArchiveSha256, - }); + })); } else { - const archive = async (cloneDir: string) => createTarballFromDirectory({ localDir: cloneDir, archivePath: gitTarPath, exclude: [".paperclip-runtime"] }); - if (gitSnapshot) await withShallowGitWorkspaceClone({ localDir: input.workspaceLocalDir, snapshot: gitSnapshot }, archive); - else await withTempDir("paperclip-managed-repositories-", async (cloneDir) => { + const archive = async (cloneDir: string) => runStepSpan("workspace.git_archive", () => createTarballFromDirectory({ localDir: cloneDir, archivePath: gitTarPath, exclude: [".paperclip-runtime"] })); + if (gitSnapshot) await runStepSpan("workspace.git_clone", () => withShallowGitWorkspaceClone({ localDir: input.workspaceLocalDir, snapshot: gitSnapshot }, archive)); + else await runStepSpan("workspace.git_clone", () => withTempDir("paperclip-managed-repositories-", async (cloneDir) => { for (const repo of repositories) await withShallowGitWorkspaceClone({ localDir: path.join(input.workspaceLocalDir, repo.path), snapshot: repo.snapshot }, async (child) => copyCloneTree(child, path.join(cloneDir, repo.path))); await archive(cloneDir); - }); + })); if (input.workspaceDurableSeed?.gitArchivePath) { - await persistDurableSeedArchive({ + await runStepSpan("workspace.seed_persist", () => persistDurableSeedArchive({ sourcePath: gitTarPath, - targetPath: input.workspaceDurableSeed.gitArchivePath, - }); + targetPath: input.workspaceDurableSeed!.gitArchivePath!, + })); } } - if (cachedSeed && input.workspaceDurableSeed?.gitArchivePath) await persistDurableSeedArchive({ sourcePath: gitTarPath, targetPath: input.workspaceDurableSeed.gitArchivePath }); + if (cachedSeed && input.workspaceDurableSeed?.gitArchivePath) await runStepSpan("workspace.seed_persist", () => persistDurableSeedArchive({ sourcePath: gitTarPath, targetPath: input.workspaceDurableSeed!.gitArchivePath! })); workspaceFiles.push({ sourcePath: gitTarPath, targetPath: remoteGitTar, @@ -1573,41 +1575,41 @@ export async function prepareSandboxManagedRuntime(input: { // 2. workspace-overlay tar. A git-backed overlay merges on top of the just // extracted git tree (no wipe); a plain workspace wipes every child except // the preserved names first. The extract runs AFTER the git extract. - await emitRuntimeStatus(input.onRuntimeProgress, "config_sync", "Syncing workspace to environment"); + await reportWorkspaceStatus("config_sync", "Syncing workspace to environment"); const workspaceTarPath = path.join(tempDir, "workspace.tar"); if (replaySeed) { - await copyDurableSeedArchive({ + await runStepSpan("workspace.seed_replay", () => copyDurableSeedArchive({ sourcePath: replaySeed.workspaceArchivePath, targetPath: workspaceTarPath, expectedSha256: replaySeed.workspaceArchiveSha256, - }); + })); } else { const workspaceArchiveDir = gitSnapshot || repositories.length ? path.join(tempDir, "workspace-overlay") : input.workspaceLocalDir; if (gitSnapshot || repositories.length) { - await copySelectedWorkspaceEntries({ + await runStepSpan("workspace.overlay_copy", () => copySelectedWorkspaceEntries({ sourceRoot: workspaceRoot!, targetDir: workspaceArchiveDir, relativePaths: (input.workspaceFileMode === "all" ? undefined : gitSnapshot?.overlayPaths) ?? (baselineSnapshot!.entries instanceof WorkspaceManifestMap ? baselineSnapshot!.entries.manifest : [...baselineSnapshot!.entries.keys()]), ignoredPaths, exclude: workspaceArchiveExclude, - }); + })); } - await createTarballFromDirectory({ + await runStepSpan("workspace.overlay_archive", () => createTarballFromDirectory({ localDir: workspaceArchiveDir, archivePath: workspaceTarPath, exclude: gitSnapshot || repositories.length ? undefined : workspaceArchiveExclude, - }); + })); if (input.workspaceDurableSeed) { - await persistDurableSeedArchive({ + await runStepSpan("workspace.seed_persist", () => persistDurableSeedArchive({ sourcePath: workspaceTarPath, - targetPath: input.workspaceDurableSeed.workspaceArchivePath, - }); + targetPath: input.workspaceDurableSeed!.workspaceArchivePath, + })); } } - if (cachedSeed && input.workspaceDurableSeed) await persistDurableSeedArchive({ sourcePath: workspaceTarPath, targetPath: input.workspaceDurableSeed.workspaceArchivePath }); + if (cachedSeed && input.workspaceDurableSeed) await runStepSpan("workspace.seed_persist", () => persistDurableSeedArchive({ sourcePath: workspaceTarPath, targetPath: input.workspaceDurableSeed!.workspaceArchivePath })); const remoteWorkspaceTar = path.posix.join( runtimeRootDir, "workspace-upload.tar", @@ -1631,7 +1633,7 @@ export async function prepareSandboxManagedRuntime(input: { if ((isPathManifest(deletedPaths) ? deletedPaths.count : deletedPaths.length) > 0) { const deletedManifest = path.join(tempDir, "deleted.nul"); const remoteDeletedManifest = path.posix.join(runtimeRootDir, "deleted.nul"); - await writeWorkspacePaths(deletedPaths, deletedManifest); + await runStepSpan("workspace.deleted_manifest", () => writeWorkspacePaths(deletedPaths, deletedManifest)); workspaceFiles.push({ sourcePath: deletedManifest, targetPath: remoteDeletedManifest, kind: "file", access: "rw", writablePath: runtimeRootDir }); workspacePostUploadCommands.push({ command: buildRemoveDeletedPathsCommand({ remoteDir: workspaceRemoteDir, manifestPath: remoteDeletedManifest }) }); } @@ -1640,14 +1642,14 @@ export async function prepareSandboxManagedRuntime(input: { if (seedGeneration && !cachedSeed && input.workspaceDurableSeed) { const seed = input.workspaceDurableSeed; - await publishWorkspaceSeedGeneration(input.workspaceSeedCacheDirectory!, seedGeneration, + await runStepSpan("workspace.seed_publish", () => publishWorkspaceSeedGeneration(input.workspaceSeedCacheDirectory!, seedGeneration, { ...seed, gitArchivePath: gitSnapshot || repositories.length ? seed.gitArchivePath : null }, { companyDirectory: input.workspaceSeedCacheCompanyDirectory, verify: () => runStepSpan("seed.verify", async () => { const verifiedRoot = path.join(tempDir, "verify-seed"); try { - if (gitSnapshot || repositories.length) await extractTarballToDirectory({ archivePath: path.join(tempDir, "git-workspace.tar"), localDir: verifiedRoot }); - await extractTarballToDirectory({ archivePath: path.join(tempDir, "workspace.tar"), localDir: verifiedRoot, exclude: repositoryMetadataExcludes }); + if (gitSnapshot || repositories.length) await runStepSpan("workspace.seed_extract", () => extractTarballToDirectory({ archivePath: path.join(tempDir, "git-workspace.tar"), localDir: verifiedRoot })); + await runStepSpan("workspace.seed_extract", () => extractTarballToDirectory({ archivePath: path.join(tempDir, "workspace.tar"), localDir: verifiedRoot, exclude: repositoryMetadataExcludes })); for (const relative of workspacePaths(deletedPaths)) { let parent = verifiedRoot; for (const segment of relative.split("/").slice(0, -1)) { @@ -1657,12 +1659,12 @@ export async function prepareSandboxManagedRuntime(input: { } await fs.rm(path.join(verifiedRoot, relative), { recursive: true, force: true }); } - const verified = await captureWorkspaceSnapshot(verifiedRoot, { exclude: baselineSnapshot!.exclude, ignoredPaths: baselineSnapshot!.ignoredPaths, diskBacked: true }); + const verified = await runStepSpan("workspace.seed_snapshot", () => captureWorkspaceSnapshot(verifiedRoot, { exclude: baselineSnapshot!.exclude, ignoredPaths: baselineSnapshot!.ignoredPaths, diskBacked: true })); try { return directorySnapshotSha256(verified) === directorySnapshotSha256(baselineSnapshot!); } finally { await disposeDirectorySnapshot(verified); } - } finally { await fs.rm(verifiedRoot, { recursive: true, force: true }); } + } finally { await runStepSpan("workspace.seed_cleanup", () => fs.rm(verifiedRoot, { recursive: true, force: true })); } }), - }); + })); } if (!stageWorkspace) return; @@ -1821,11 +1823,11 @@ export async function prepareSandboxManagedRuntime(input: { // tasks under the client's concurrency permission and the bound, and it // settles every started task before it returns. This is the startup barrier: // the coordinator returns only after every upload settles. - const inboundResults = await scheduleSyncOperations( + const inboundResults = await runStepSpan("workspace.inbound_barrier", () => scheduleSyncOperations( inboundTasks, input.client.allowConcurrentSyncOperations === true, SYNC_OPERATION_CONCURRENCY_LIMIT, - ); + )); // Collect the referenced-project failures in stable input order. for (const failure of additionalFailureSlots) { diff --git a/server/src/services/chat-channels/slack/task-authority.ts b/server/src/services/chat-channels/slack/task-authority.ts index 43b5673126d..8fa8f85396b 100644 --- a/server/src/services/chat-channels/slack/task-authority.ts +++ b/server/src/services/chat-channels/slack/task-authority.ts @@ -5,6 +5,8 @@ import { slackAuthorizationRevision } from "../../connectors/slack-revision.js"; import { instanceSettingsService } from "../../instance-settings.js"; import { registerSlackTaskAuthority, slackRunOrigin } from "../../connectors/slack-authority.js"; +import { measureRunPhase } from "../../execution-preparation-timing.js"; +import { measureRunTransaction } from "../../execution-transaction-timing.js"; import { createHash, @@ -51,13 +53,13 @@ export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskA const unregisterSlackTaskAuthority = registerSlackTaskAuthority(db, async (binding) => { - if (!(await instanceSettingsService(db).getExperimental()).enableChatConnectors) + if (!(await measureRunPhase("connectors.slack.settings", () => instanceSettingsService(db).getExperimental())).enableChatConnectors) throw forbidden("Chat connectors are disabled"); - const identity = await slackRunOrigin(db, binding); + const identity = await measureRunPhase("connectors.slack.identity", () => slackRunOrigin(db, binding)); const run = identity.run; if ((run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId) !== binding.issueId || !run.responsibleUserId) throw forbidden("Slack tools require the bound task and an accepted linked user"); - const resolved = await db.transaction(async (tx) => { + const resolved = await measureRunTransaction(db, "connectors.slack.authority", async (tx) => { const [action] = await tx.select().from(chatActions).where(and( eq(chatActions.companyId, binding.companyId), eq(chatActions.kind, "inbound_wakeup"), @@ -134,7 +136,7 @@ export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskA if (!principal) throw forbidden("Slack sender is no longer available"); return { ...source, principalId: principal.id, slackUserId: principal.externalId }; }); - const credentials = await resolveCredentials(resolved.endpoint); + const credentials = await measureRunPhase("connectors.slack.credentials", () => resolveCredentials(resolved.endpoint)); if (!credentials.botToken) throw forbidden("Slack bot credential is unavailable"); return { endpoint: resolved.endpoint, issueId: binding.issueId, conversation: resolved.conversation, @@ -145,7 +147,7 @@ export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskA revision: createHash("sha256").update(JSON.stringify([ resolved.endpoint.status, resolved.endpoint.allowDirectMessages, resolved.conversation?.sessionGeneration ?? null, run.responsibleUserId, credentials.botToken, - await slackAuthorizationRevision(db, binding.companyId, resolved.endpoint.id, resolved.endpoint.connectionId, binding.agentId, run.responsibleUserId), + await measureRunPhase("connectors.slack.revision", () => slackAuthorizationRevision(db, binding.companyId, resolved.endpoint.id, resolved.endpoint.connectionId, binding.agentId, run.responsibleUserId!)), ])).digest("hex"), botToken: credentials.botToken, searchActionToken: resolved.delivery ? slackSearchActionToken(db, resolved.endpoint.id, resolved.endpoint.providerAccountId!, resolved.slackUserId, String((resolved.delivery.normalizedEvent.message as Record | undefined)?.providerMessageId ?? "")) : null, diff --git a/server/src/services/connector-runtime.ts b/server/src/services/connector-runtime.ts index 4ccd1bb89bb..6ab346660f8 100644 --- a/server/src/services/connector-runtime.ts +++ b/server/src/services/connector-runtime.ts @@ -22,6 +22,7 @@ import { } from "./connectors/agentmail.js"; import { materializeAsset } from "./native-runtime/runtime-context.js"; import { BROWSER_USE_CLOUD_SKILL } from "./connectors/browser-use-cloud/skill.js"; +import { measureRunPhase, type ExecutionPhase } from "./execution-preparation-timing.js"; type AgentBinding = { companyId: string; agentId: string; runId?: string; issueId?: string }; type ToolBinding = AgentBinding & { @@ -43,6 +44,7 @@ type Tool = { }; interface ConnectorDefinition { key: string; + resolvePhase: Extract; label: string; skillName: string; /** Trusted connector-owned content; never discovered through global skills/. */ @@ -62,7 +64,7 @@ interface ConnectorDefinition { // Trusted connector packages declare their contributions here. Assignments and // current access, not credential availability or agent-authored config, select them. const connectors: ConnectorDefinition[] = [ - { key: "browser-use-cloud", label: "Browser Use Cloud", skillName: "browser-use-cloud", skillMarkdown: BROWSER_USE_CLOUD_SKILL, tools: [], + { key: "browser-use-cloud", resolvePhase: "connectors.browser_use.resolve", label: "Browser Use Cloud", skillName: "browser-use-cloud", skillMarkdown: BROWSER_USE_CLOUD_SKILL, tools: [], legacySkillKeys: ["paperclipai/paperclip/browser-use"], async resolve(db, binding) { const { getAssignedMcpGateway } = await import("./native-runtime/assigned-mcp-tools.js"); @@ -70,18 +72,19 @@ const connectors: ConnectorDefinition[] = [ }, async execute() { throw forbidden("Use Browser Use Cloud through the connection tool gateway."); }, }, - { key: "speko", label: "Speko", skillName: "speko", skillMarkdown: SPEKO_SKILL, + { key: "speko", resolvePhase: "connectors.speko.resolve", label: "Speko", skillName: "speko", skillMarkdown: SPEKO_SKILL, tools: [], // Native phone tools keep their existing governed gateway transport. resolve: spekoAssignedResources, async execute() { throw forbidden("Use Speko through the connection tool gateway."); }, }, - { key: "slack", label: "Slack", skillName: "slack", + { key: "slack", resolvePhase: "connectors.slack.resolve", label: "Slack", skillName: "slack", tools: SLACK_TOOLS.map(({ name, description, inputSchema }) => ({ name, description, inputSchema })), resolve: slackAssignedResource, execute: executeGovernedSlackTool, }, { key: "agentmail", + resolvePhase: "connectors.agentmail.resolve", label: "AgentMail", skillName: "agentmail", tools: AGENTMAIL_TOOLS.map( @@ -127,7 +130,7 @@ export async function resolveConnectorAssignments( ): Promise { const assignments: ConnectorAssignment[] = []; for (const connector of connectors) { - const resources = await connector.resolve(db, binding); + const resources = await measureRunPhase(connector.resolvePhase, () => connector.resolve(db, binding)); if (resources.length) assignments.push({ key: connector.key, diff --git a/server/src/services/connectors/slack-catalog.ts b/server/src/services/connectors/slack-catalog.ts index 607c6c44e8e..184695da3b2 100644 --- a/server/src/services/connectors/slack-catalog.ts +++ b/server/src/services/connectors/slack-catalog.ts @@ -1,4 +1,5 @@ import { createHash } from "node:crypto"; +import { measureRunPhase } from "../execution-preparation-timing.js"; import { and, eq, isNull, sql } from "drizzle-orm"; import { SLACK_TOOLS } from "@paperclipai/shared"; import { @@ -35,13 +36,13 @@ export async function syncSlackBotTools( ), ); if (!connection) return; - await tx.execute( + await measureRunPhase("connectors.slack.catalog.lock", () => tx.execute( sql`select pg_advisory_xact_lock(hashtextextended(${`slack-catalog:${endpoint.id}`}, 0))`, - ); - const existingEntries = await tx + )); + const existingEntries = await measureRunPhase("connectors.slack.catalog.entries", () => tx .select() .from(toolCatalogEntries) - .where(eq(toolCatalogEntries.connectionId, connection.id)); + .where(eq(toolCatalogEntries.connectionId, connection.id))); const entriesByName = new Map( existingEntries.map((entry) => [entry.name, entry]), ); @@ -79,7 +80,7 @@ export async function syncSlackBotTools( }; const existing = entriesByName.get(values.name); if (existing?.versionHash === hash) continue; - await tx + await measureRunPhase("connectors.slack.catalog.write", () => tx .insert(toolCatalogEntries) .values(values) .onConflictDoUpdate({ @@ -94,10 +95,10 @@ export async function syncSlackBotTools( } : {}), }, - }); + })); } const profileKey = `slack-bot:${endpoint.id}`; - const [priorProfile] = await tx + const [priorProfile] = await measureRunPhase("connectors.slack.catalog.profile", () => tx .select() .from(toolProfiles) .where( @@ -105,7 +106,7 @@ export async function syncSlackBotTools( eq(toolProfiles.companyId, endpoint.companyId), eq(toolProfiles.profileKey, profileKey), ), - ); + )); if (priorProfile) return; const [profile] = await tx .insert(toolProfiles) diff --git a/server/src/services/connectors/slack.ts b/server/src/services/connectors/slack.ts index 9cbde7bc034..3053feb2867 100644 --- a/server/src/services/connectors/slack.ts +++ b/server/src/services/connectors/slack.ts @@ -4,6 +4,8 @@ import { slackMessage } from "./slack-message.js"; import { executeSlackWrite } from "./slack-writes.js"; import { syncSlackBotTools } from "./slack-catalog.js"; import { SLACK_TOOLS } from "@paperclipai/shared"; +import { measureRunPhase } from "../execution-preparation-timing.js"; +import { measureRunTransaction } from "../execution-transaction-timing.js"; import { toolProfiles, toolProfileBindings, @@ -36,10 +38,10 @@ async function slackAssignedEndpoint( if (!binding.runId || !binding.issueId) return []; let authority: Awaited>; try { - authority = await resolveSlackTaskAuthority( + authority = await measureRunPhase("connectors.slack.authority", () => resolveSlackTaskAuthority( db, binding as SlackTaskBinding, - ); + )); } catch (error) { if ( error && @@ -51,10 +53,10 @@ async function slackAssignedEndpoint( throw error; } // Catalog updates are idempotent and preserve configured policies. - await db.transaction((tx) => + await measureRunTransaction(db, "connectors.slack.catalog", (tx) => syncSlackBotTools(tx, authority.endpoint, authority.userId, true), ); - const [enabledProfile] = await db + const [enabledProfile] = await measureRunPhase("connectors.slack.profile", () => db .select({ id: toolProfiles.id }) .from(toolProfiles) .innerJoin( @@ -81,9 +83,9 @@ async function slackAssignedEndpoint( eq(toolProfiles.status, "active"), ), ) - .limit(1); + .limit(1)); if (!enabledProfile) return []; - authority = await resolveSlackTaskAuthority(db, binding as SlackTaskBinding); + authority = await measureRunPhase("connectors.slack.authority", () => resolveSlackTaskAuthority(db, binding as SlackTaskBinding)); return [ { id: authority.endpoint.id, @@ -106,7 +108,7 @@ export async function slackAssignedResource( binding: Partial & { companyId: string; agentId: string }, ) { const resources = []; - for (const endpoint of await slackEndpointCandidates(db, binding)) { + for (const endpoint of await measureRunPhase("connectors.slack.candidates", () => slackEndpointCandidates(db, binding))) { resources.push( ...(await slackAssignedEndpoint(db, { ...binding, diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index 58fc7a33cc5..0df68c93e30 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -14,6 +14,10 @@ type PreparationPhase = "environment_selection" | "execution_configuration" | "s | "workspace_provision_policy" | "workspace_base_validation" | "workspace_source_authorization" + | "workspace_source_materialize" + | "workspace_ownership_read" + | "workspace_environment_acquire" + | "workspace_environment_realize" | "workspace_provision" | "workspace_source_retention" | "workspace_binding" @@ -36,6 +40,9 @@ export type ExecutionPhase = | "workspace.sources.repository_scan" | "workspace.sources.project_scan" | "workspace.provenance_lock" | "workspace.readers" | "workspace.readers.identity" | "workspace.readers.trust" | "workspace.readers.projects" | "workspace.readers.sources" | "workspace.access" | "workspace.access.privacy" | "workspace.access.inventory" | "workspace.access.authorization" + | "workspace.materialize.directories" | "workspace.materialize.source_identity" + | "workspace.checkout.lookup" | "workspace.checkout.git_auth" | "workspace.checkout.git_clone" + | "workspace.checkout.git_ref" | "workspace.checkout.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" @@ -44,6 +51,26 @@ export type ExecutionPhase = | "runtime_context.instructions" | "runtime_context.skills" | "runtime_context.mcp" | "runtime_context.asset_read" | "runtime_context.asset_manifest" | "runtime_context.asset_cache" | "runtime_context.asset_materialize" | "runtime_context.asset_verify" | "workspace_sync.snapshot_git" | "workspace_sync.snapshot_baseline" | "workspace_sync.pack" | "workspace_sync.seed_verify" | "workspace_sync.stage_workspace" | "workspace_sync.restore" + | "workspace_sync.run_read" | "workspace_sync.descriptor_directory" | "workspace_sync.descriptor_read" + | "workspace_sync.prepare_host_current" | "workspace_sync.prepare_durable_seed" | "workspace_sync.prepare_adopt_remote" + | "workspace_sync.archive_hash" | "workspace_sync.warm_stamp" | "workspace_sync.durable_seed_verify" + | "workspace_sync.manifests_persist" | "workspace_sync.descriptor_write" | "workspace_sync.reference_write" + | "workspace_sync.source_root" | "workspace_sync.repositories" | "workspace_sync.ignore_rules" + | "workspace_sync.seed_generation" | "workspace_sync.seed_lookup" | "workspace_sync.seed_publish" + | "workspace_sync.git_clone" | "workspace_sync.git_archive" | "workspace_sync.seed_replay" | "workspace_sync.seed_persist" + | "workspace_sync.overlay_copy" | "workspace_sync.overlay_archive" | "workspace_sync.deleted_manifest" + | "workspace_sync.seed_extract" | "workspace_sync.seed_snapshot" | "workspace_sync.seed_cleanup" + | "workspace_sync.inbound_barrier" | "workspace_sync.transfer" | "workspace_sync.progress" + | "connectors.browser_use.resolve" | "connectors.speko.resolve" | "connectors.slack.resolve" | "connectors.agentmail.resolve" + | "connectors.browser_use.connection" | "connectors.browser_use.catalog" | "connectors.browser_use.grant" + | "connectors.discovery.scheduled" | "connectors.discovery.queue" | "connectors.discovery.build" | "connectors.discovery.agent" + | "connectors.discovery.connections" | "connectors.discovery.github" | "connectors.discovery.slack" | "connectors.discovery.speko" + | "connectors.discovery.policy" | "connectors.discovery.on_demand_policy" + | "connectors.slack.candidates" | "connectors.slack.authority" | "connectors.slack.profile" + | "connectors.slack.settings" | "connectors.slack.identity" | "connectors.slack.credentials" | "connectors.slack.revision" + | "connectors.slack.authority.transaction" | "connectors.slack.authority.transaction.entry" | "connectors.slack.authority.transaction.commit" | "connectors.slack.authority.transaction.rollback" + | "connectors.slack.catalog.transaction" | "connectors.slack.catalog.transaction.entry" | "connectors.slack.catalog.transaction.commit" | "connectors.slack.catalog.transaction.rollback" + | "connectors.slack.catalog.lock" | "connectors.slack.catalog.entries" | "connectors.slack.catalog.write" | "connectors.slack.catalog.profile" | "execution.runtime" | "execution.workspace_ownership" | "execution.workspace_finalize" | "execution.workspace_restore" | "execution.workspace_commit" | "execution.workspace_ownership_wait" | "execution.realtime" | "execution.usage" | "execution.native_finalize" @@ -138,16 +165,58 @@ export function beginRunPhase(phase: ExecutionPhase): (failed?: boolean) => void } catch { return () => {}; } } +export function measureToolDiscovery(schedule: (work: () => Promise) => Promise, work: () => Promise): Promise { + return measureRunPhase("connectors.discovery.scheduled", async () => { + const queued = beginRunPhase("connectors.discovery.queue"); + const build = AsyncLocalStorage.bind(() => { + queued(); + return measureRunPhase("connectors.discovery.build", work); + }); + let failed = false; + try { return await schedule(build); } + catch (error) { failed = true; throw error; } + finally { queued(failed); } + }); +} + export function createRunPreparationTiming(runId: string, scope: "heartbeat.prepare" | "native.controller" = "heartbeat.prepare") { return (phase: PreparationPhase, work: () => T | Promise): Promise => measure(runId, `${scope}.${phase}`, work); } +const workspaceSyncDetailPhases = { + "workspace.source_root": "workspace_sync.source_root", + "workspace.repositories": "workspace_sync.repositories", + "workspace.ignore_rules": "workspace_sync.ignore_rules", + "workspace.seed_generation": "workspace_sync.seed_generation", + "workspace.seed_lookup": "workspace_sync.seed_lookup", + "workspace.seed_publish": "workspace_sync.seed_publish", + "workspace.git_clone": "workspace_sync.git_clone", + "workspace.git_archive": "workspace_sync.git_archive", + "workspace.seed_replay": "workspace_sync.seed_replay", + "workspace.seed_persist": "workspace_sync.seed_persist", + "workspace.overlay_copy": "workspace_sync.overlay_copy", + "workspace.overlay_archive": "workspace_sync.overlay_archive", + "workspace.deleted_manifest": "workspace_sync.deleted_manifest", + "workspace.seed_extract": "workspace_sync.seed_extract", + "workspace.seed_snapshot": "workspace_sync.seed_snapshot", + "workspace.seed_cleanup": "workspace_sync.seed_cleanup", + "workspace.inbound_barrier": "workspace_sync.inbound_barrier", + "workspace.transfer": "workspace_sync.transfer", + "workspace.progress": "workspace_sync.progress", +} as const satisfies Record; + +export function isWorkspaceSyncDetailPhase(name: string): boolean { + return Object.prototype.hasOwnProperty.call(workspaceSyncDetailPhases, name); +} + +const workspaceSyncPhases: Readonly> = { + ...workspaceSyncDetailPhases, + "snapshot.git": "workspace_sync.snapshot_git", "snapshot.baseline": "workspace_sync.snapshot_baseline", + pack: "workspace_sync.pack", "seed.verify": "workspace_sync.seed_verify", + "stage.workspace": "workspace_sync.stage_workspace", "restore.workspace": "workspace_sync.restore", + "workspace.restore": "execution.workspace_restore", +}; + export function measureWorkspaceSyncPhase(name: string, work: () => Promise): Promise { - const phases: Readonly> = { - "snapshot.git": "workspace_sync.snapshot_git", "snapshot.baseline": "workspace_sync.snapshot_baseline", - pack: "workspace_sync.pack", "seed.verify": "workspace_sync.seed_verify", - "stage.workspace": "workspace_sync.stage_workspace", "restore.workspace": "workspace_sync.restore", - "workspace.restore": "execution.workspace_restore", - }; - return Object.prototype.hasOwnProperty.call(phases, name) ? measureRunPhase(phases[name]!, work) : work(); + return Object.prototype.hasOwnProperty.call(workspaceSyncPhases, name) ? measureRunPhase(workspaceSyncPhases[name]!, work) : work(); } diff --git a/server/src/services/execution-timing-context.test.ts b/server/src/services/execution-timing-context.test.ts index 864b07ccf43..7ac9ca1f6b1 100644 --- a/server/src/services/execution-timing-context.test.ts +++ b/server/src/services/execution-timing-context.test.ts @@ -7,7 +7,8 @@ import { afterEach, beforeEach, expect, it, vi } from "vitest"; import { getStartupTraceContext, getStartupTracer } from "../instrumentation.js"; import { prepareVerifiedRemoteProviderPack } from "./native-runtime/remote-provider-pack.js"; import { createNativeRunTrace } from "./native-runtime/native-run-trace.js"; -import { beginRunPhase, createRunPreparationTiming, getRunExecutionTraceContext, measureRunPhase, withRunExecutionTiming } from "./execution-preparation-timing.js"; +import { beginRunPhase, createRunPreparationTiming, getRunExecutionTraceContext, measureRunPhase, measureToolDiscovery, withRunExecutionTiming } from "./execution-preparation-timing.js"; +import { createToolDiscoveryScheduler } from "./tool-discovery-scheduler.js"; let exporter: InMemorySpanExporter; let provider: TracerProvider; @@ -92,6 +93,54 @@ it("uses a link for a native session whose queue history predates the current at expect(session.startTime[0]).toBeLessThan(attempt.startTime[0]); }); +it("keeps queued discovery under its initiating run when another run releases the slot", async () => { + const scheduler = createToolDiscoveryScheduler(1, 1); + let release!: () => void, started!: () => void, enqueued!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { started = resolve; }); + const waiting = new Promise(resolve => { enqueued = resolve; }); + const first = withRunExecutionTiming("discovery-owner", () => measureToolDiscovery( + work => scheduler.run(work), async () => { started(); await gate; }, + )); + await ready; + const runHash = createHash("sha256").update("discovery-waiter").digest("hex").slice(0, 12); + const value = {}; + const work = vi.fn(async () => { + expect(getRunExecutionTraceContext("discovery-owner")).toBeUndefined(); + expect(getRunExecutionTraceContext("discovery-waiter")).toBeDefined(); + expect(exporter.getFinishedSpans().filter(span => span.name === "connectors.discovery.queue" + && span.attributes["paperclip.task.run.run_id"] === runHash)).toHaveLength(1); + await measureRunPhase("connectors.discovery.policy", () => undefined); + return value; + }); + const second = withRunExecutionTiming("discovery-waiter", () => measureToolDiscovery( + work => { const pending = scheduler.run(work); enqueued(); return pending; }, work, + )); + await waiting; + try { + expect(work).not.toHaveBeenCalled(); + expect(exporter.getFinishedSpans().some(span => span.attributes["paperclip.task.run.run_id"] === runHash)).toBe(false); + } finally { + release(); + await first; + } + await expect(second).resolves.toBe(value); + expect(work).toHaveBeenCalledOnce(); + const spans = exporter.getFinishedSpans(); + const owned = spans.filter(span => span.attributes["paperclip.task.run.run_id"] === runHash); + const attempt = owned.find(span => span.name === "task.execution_attempt")!; + const scheduled = owned.find(span => span.name === "connectors.discovery.scheduled")!; + const build = owned.find(span => span.name === "connectors.discovery.build")!; + const queue = owned.filter(span => span.name === "connectors.discovery.queue"); + expect(queue).toHaveLength(1); + expect(owned).toHaveLength(5); + expect(owned.every(span => span.spanContext().traceId === attempt.spanContext().traceId)).toBe(true); + expect(scheduled.parentSpanContext?.spanId).toBe(attempt.spanContext().spanId); + expect(build.parentSpanContext?.spanId).toBe(scheduled.spanContext().spanId); + expect(queue[0].parentSpanContext?.spanId).toBe(scheduled.spanContext().spanId); + expect(owned.find(span => span.name === "connectors.discovery.policy")?.parentSpanContext?.spanId).toBe(build.spanContext().spanId); +}); + it("caps repeated detail spans and ends failed scopes without content or exception text", async () => { const failure = new Error("private path and credential"); const work = vi.fn(() => undefined); diff --git a/server/src/services/execution-transaction-timing.ts b/server/src/services/execution-transaction-timing.ts index bb22c861be0..1cf4583fcd9 100644 --- a/server/src/services/execution-transaction-timing.ts +++ b/server/src/services/execution-transaction-timing.ts @@ -3,7 +3,7 @@ import { beginRunPhase, measureRunPhase } from "./execution-preparation-timing.j type Transaction = Parameters[0]>[0]; -export function measureRunTransaction(db: Db, scope: "workspace" | "runtime_selection" | "response", work: (tx: Transaction) => Promise): Promise { +export function measureRunTransaction(db: Db, scope: "workspace" | "runtime_selection" | "response" | "connectors.slack.authority" | "connectors.slack.catalog", work: (tx: Transaction) => Promise): Promise { return measureRunPhase(`${scope}.transaction`, async () => { const entry = beginRunPhase(`${scope}.transaction.entry`); let settle: ((failed?: boolean) => void) | undefined; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9cd1ff8635e..221ed1632f5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,5 +1,5 @@ import { dotRunnerBroker } from "./dot-runner-broker.js"; -import { withRunExecutionTiming, measureRunPhase, measureWorkspaceSyncPhase } from "./execution-preparation-timing.js"; +import { withRunExecutionTiming, measureRunPhase, measureWorkspaceSyncPhase, isWorkspaceSyncDetailPhase } from "./execution-preparation-timing.js"; import { createHeartbeatPreparationTiming } from "./heartbeat/preparation-timing.js"; import { createHeartbeatInstructionCopyRelease, createHeartbeatInstructionWorkingCopy } from "./heartbeat/instruction-working-copy.js"; import { prepareHeartbeatRuntimeServiceIntents, startHeartbeatRuntimeServices, recordHeartbeatAdapterRuntimeServices } from "./heartbeat/runtime-services.js"; @@ -3320,10 +3320,11 @@ export function heartbeatService( message: "workspace_checkpoint", payload: { phase: "workspace_checkpoint", ...metrics } }); }, onPhase: async (name, work) => { + if (isWorkspaceSyncDetailPhase(name)) return measureWorkspaceSyncPhase(name, work); const started = Date.now(); try { return await measureWorkspaceSyncPhase(name, work); } - finally { await appendRunEvent(run, { eventType: "lifecycle", stream: "system", level: "info", - message: name, payload: { phase: name, durationMs: Date.now() - started } }); } + finally { await measureRunPhase("workspace_sync.progress", () => appendRunEvent(run, { eventType: "lifecycle", stream: "system", level: "info", + message: name, payload: { phase: name, durationMs: Date.now() - started } })); } }, }, config: { diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index 5969afb6ffb..92cb041602a 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -508,8 +508,8 @@ export async function prepareHeartbeatWorkspace(db: Db, input: HeartbeatWorkspac selectedProjectWorkspaceId: sourceChoice.projectWorkspaceId ?? readNonEmptyString(context.projectWorkspaceId), selection: sourceChoice.selection, strategy: parseObject(hostExecutionWorkspaceConfig.workspaceStrategy), trustPresetKind: trustPreset.kind })) throw error; - const cwd = await materializeIsolatedRemoteWorkspaceSource({ companyId: agent.companyId, - issueId, repoUrl: executionWorkspaceBase.repoUrl!, resolveGitAuth: workspaceGitAuthProvider }); + const cwd = await timeWorkspace("workspace_source_materialize", () => materializeIsolatedRemoteWorkspaceSource({ companyId: agent.companyId, + issueId, repoUrl: executionWorkspaceBase.repoUrl!, resolveGitAuth: workspaceGitAuthProvider })); executionWorkspaceBase.baseCwd = cwd; resolvedWorkspace.cwd = cwd; isolatedRemoteSource = { schema: "paperclip.isolated-remote-source.v1", companyId: agent.companyId, @@ -762,9 +762,9 @@ export async function prepareHeartbeatWorkspace(db: Db, input: HeartbeatWorkspac try { persistedWorktreeInstanceRoot = ( - await readManagedWorktreeInstanceOwnership( - executionWorkspace.worktreePath, - ) + await timeWorkspace("workspace_ownership_read", () => readManagedWorktreeInstanceOwnership( + executionWorkspace.worktreePath!, + )) )?.instanceRoot ?? null; } catch (error) { logger.warn( @@ -1007,7 +1007,7 @@ export async function prepareHeartbeatWorkspace(db: Db, input: HeartbeatWorkspac recoveryWorkspace.remoteCwd !== remoteRecovery.remote.remoteCwd)) { throw new Error("native_remote_recovery_lease_mismatch"); } - acquiredEnvironment = await envOrchestrator.acquireForRun({ + acquiredEnvironment = await timeWorkspace("workspace_environment_acquire", () => envOrchestrator.acquireForRun({ companyId: agent.companyId, selectedEnvironmentId, localEnvironmentId: localEnvironment.id, @@ -1024,7 +1024,7 @@ export async function prepareHeartbeatWorkspace(db: Db, input: HeartbeatWorkspac providerLeaseId: remoteRecovery.remote.providerLeaseId, remoteCwd: remoteRecovery.remote.remoteCwd, } } : {}), - }); + })); await controllerLease.assertOwned(); nativeRunnerPreparationSpans.push({ name: "environment.acquire", @@ -1100,7 +1100,7 @@ export async function prepareHeartbeatWorkspace(db: Db, input: HeartbeatWorkspac ReturnType >; try { - realizationResult = await envOrchestrator.realizeForRun({ + realizationResult = await timeWorkspace("workspace_environment_realize", () => envOrchestrator.realizeForRun({ environment: selectedEnvironment, lease: activeEnvironmentLease.lease, adapterType: agent.adapterType, @@ -1111,7 +1111,7 @@ export async function prepareHeartbeatWorkspace(db: Db, input: HeartbeatWorkspac effectiveExecutionWorkspaceMode, persistedExecutionWorkspace, duplexObservabilityRecorder, - }); + })); nativeRunnerPreparationSpans.push({ name: "environment.workspace.realize", parentName: "task.provider_session", diff --git a/server/src/services/isolated-remote-workspace-source.ts b/server/src/services/isolated-remote-workspace-source.ts index 26c949fdfbc..efb299db2e1 100644 --- a/server/src/services/isolated-remote-workspace-source.ts +++ b/server/src/services/isolated-remote-workspace-source.ts @@ -6,6 +6,7 @@ import type { TaskWorkspaceIntent } from "@paperclipai/shared"; import type { GitRemoteAuthProvider } from "./git-credentials.js"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { materializeManagedProjectWorkspace } from "./managed-repository-checkout.js"; +import { measureRunPhase } from "./execution-preparation-timing.js"; const execFile = promisify(execFileCallback); @@ -89,32 +90,35 @@ export async function materializeIsolatedRemoteWorkspaceSource(input: { resolveGitAuth: GitRemoteAuthProvider; }): Promise { const cwd = isolatedRemoteSourcePath(input.companyId, input.issueId); - const root = resolvePaperclipInstanceRoot(); - await fs.mkdir(root, { recursive: true }); - const canonicalRoot = await fs.realpath(root); - let parent = canonicalRoot; - for (const segment of ["isolated-repository-sources", input.companyId, input.issueId]) { - parent = path.join(parent, segment); - await fs.mkdir(parent, { mode: 0o700 }).catch((error: NodeJS.ErrnoException) => { - if (error.code !== "EEXIST") throw error; - }); - const entry = await fs.lstat(parent); - if (!entry.isDirectory() || entry.isSymbolicLink() || await fs.realpath(parent) !== parent) { - throw new Error("Isolated repository source path is not a private directory"); + const canonicalCwd = await measureRunPhase("workspace.materialize.directories", async () => { + const root = resolvePaperclipInstanceRoot(); + await fs.mkdir(root, { recursive: true }); + const canonicalRoot = await fs.realpath(root); + let parent = canonicalRoot; + for (const segment of ["isolated-repository-sources", input.companyId, input.issueId]) { + parent = path.join(parent, segment); + await fs.mkdir(parent, { mode: 0o700 }).catch((error: NodeJS.ErrnoException) => { + if (error.code !== "EEXIST") throw error; + }); + const entry = await fs.lstat(parent); + if (!entry.isDirectory() || entry.isSymbolicLink() || await fs.realpath(parent) !== parent) { + throw new Error("Isolated repository source path is not a private directory"); + } } - } - const canonicalCwd = path.join(parent, "checkout"); - const entry = await fs.lstat(canonicalCwd).catch((error: NodeJS.ErrnoException) => { - if (error.code === "ENOENT") return null; - throw error; + const checkout = path.join(parent, "checkout"); + const entry = await fs.lstat(checkout).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (entry && (!entry.isDirectory() || entry.isSymbolicLink())) { + throw new Error("Isolated repository source cannot be a symbolic link"); + } + return checkout; }); - if (entry && (!entry.isDirectory() || entry.isSymbolicLink())) { - throw new Error("Isolated repository source cannot be a symbolic link"); - } const result = await materializeManagedProjectWorkspace(canonicalCwd, { repoUrl: input.repoUrl, resolveGitAuth: input.resolveGitAuth, }); if (result.warning) throw new Error("Isolated repository source is not a Git checkout"); - await assertRemoteSourceIdentity(cwd, input.repoUrl); + await measureRunPhase("workspace.materialize.source_identity", () => assertRemoteSourceIdentity(cwd, input.repoUrl)); return cwd; } diff --git a/server/src/services/managed-repository-checkout.ts b/server/src/services/managed-repository-checkout.ts index f8810406f06..e8316869473 100644 --- a/server/src/services/managed-repository-checkout.ts +++ b/server/src/services/managed-repository-checkout.ts @@ -8,6 +8,7 @@ import { sanitizeRuntimeServiceBaseEnv } from "./runtime-service-env.js"; import { describeGitAuthFailure, scrubGitCredentialText, type GitRemoteAuthProvider } from "./git-credentials.js"; import { classifyGitCloneFailure, GitConnectionFailureError, readGitConnectionFailure } from "./git-connection-failure.js"; import { isWorkspaceGitScanError, WorkspaceGitScanError } from "./workspace-git-operation-scheduler.js"; +import { measureRunPhase } from "./execution-preparation-timing.js"; const execFile = promisify(execFileCallback); const MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS = 10 * 60 * 1000; @@ -34,11 +35,11 @@ export async function materializeManagedProjectWorkspace( return { cwd, warning: null }; } - const hasAdoptableGitDir = () => + const hasAdoptableGitDir = () => measureRunPhase("workspace.checkout.lookup", () => fs .stat(path.resolve(cwd, ".git")) .then((entry) => entry.isDirectory()) - .catch(() => false); + .catch(() => false)); if (await hasAdoptableGitDir()) { return { cwd, warning: null }; } @@ -55,17 +56,18 @@ export async function materializeManagedProjectWorkspace( } const auth = input.resolveGitAuth && !input.localSource - ? await input.resolveGitAuth(input.repoUrl) + ? await measureRunPhase("workspace.checkout.git_auth", () => input.resolveGitAuth!(input.repoUrl!)) : null; const cloneTmpDir = await fs.mkdtemp(input.stagingParent ? path.join(input.stagingParent, `${path.basename(cwd)}.clone-`) : `${cwd}.clone-`); + const cloneSource = input.localSource ?? input.repoUrl; try { await input.beforeCopy?.(cloneTmpDir); try { - await execFile( + await measureRunPhase("workspace.checkout.git_clone", () => execFile( "git", - [...(auth?.configArgs ?? []), "clone", "--no-hardlinks", "--", input.localSource ?? input.repoUrl, cloneTmpDir], + [...(auth?.configArgs ?? []), "clone", "--no-hardlinks", "--", cloneSource, cloneTmpDir], { env: { ...sanitizeRuntimeServiceBaseEnv(process.env), @@ -74,7 +76,7 @@ export async function materializeManagedProjectWorkspace( }, timeout: MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS, }, - ); + )); } catch (error) { const connectionFailure = !input.localSource ? classifyGitCloneFailure(input.repoUrl, error) : null; if (connectionFailure) throw new GitConnectionFailureError( @@ -95,7 +97,7 @@ export async function materializeManagedProjectWorkspace( } await execFile("git", ["-C", cloneTmpDir, "remote", "set-url", "origin", input.repoUrl], { timeout: 10_000 }); } else if (input.repoRef) { - await execFile("git", ["-C", cloneTmpDir, "checkout", input.repoRef], { timeout: MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS }); + await measureRunPhase("workspace.checkout.git_ref", () => execFile("git", ["-C", cloneTmpDir, "checkout", input.repoRef!], { timeout: MANAGED_WORKSPACE_GIT_CLONE_TIMEOUT_MS })); } await input.beforePublish?.(cloneTmpDir); } catch (error) { @@ -117,7 +119,7 @@ export async function materializeManagedProjectWorkspace( } try { - await fs.rename(cloneTmpDir, cwd); + await measureRunPhase("workspace.checkout.publish", () => fs.rename(cloneTmpDir, cwd)); } catch (renameError) { await fs .rm(cloneTmpDir, { recursive: true, force: true }) diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index 8c1399b64fc..99fe0a61c83 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -182,13 +182,13 @@ function durableSeedPaths(runId: string): { } async function sha256File(filePath: string): Promise { - return await new Promise((resolveDigest, rejectDigest) => { + return measureRunPhase("workspace_sync.archive_hash", () => new Promise((resolveDigest, rejectDigest) => { const digest = createHash("sha256"); const stream = createReadStream(filePath); stream.on("data", (chunk) => digest.update(chunk)); stream.on("error", rejectDigest); stream.on("end", () => resolveDigest(digest.digest("hex"))); - }); + })); } async function verifiedDurableSeed(input: { @@ -795,7 +795,12 @@ async function prepareRuntime(input: { onPhase?: RuntimeSpanRunner; onCheckpoint?: (metrics: WorkspaceCheckpointMetrics) => Promise; }): Promise { - return prepareAdapterExecutionTargetRuntime({ + const phase = { + host_current: "workspace_sync.prepare_host_current", + durable_seed: "workspace_sync.prepare_durable_seed", + adopt_remote: "workspace_sync.prepare_adopt_remote", + } as const; + return measureRunPhase(phase[input.mode], () => prepareAdapterExecutionTargetRuntime({ runId: input.runId, target: input.target, adapterKey: "paperclip-runner", @@ -812,7 +817,7 @@ async function prepareRuntime(input: { workspaceSeedCacheDirectory: input.seedCacheDirectory, workspaceSeedCacheCompanyDirectory: input.seedCacheCompanyDirectory, runtimeSpan: input.onPhase, - }); + })); } async function finalizePreparedRuntime(input: { @@ -889,12 +894,12 @@ export async function prepareNativeWorkspaceSync(input: { input.restartRecovery.remote.remoteCwd !== target.remoteCwd)) { throw new Error("native_remote_recovery_lease_mismatch"); } - const run = await input.db + const run = await measureRunPhase("workspace_sync.run_read", () => input.db .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1) - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (!run) throw new Error("native_workspace_sync_run_missing"); const existingReference = readNativeWorkspaceSyncReference( parseObject(run.runnerProfileJson).nativeWorkspaceSync, @@ -904,14 +909,14 @@ export async function prepareNativeWorkspaceSync(input: { let descriptor: NativeWorkspaceSyncDescriptor; let mode: WorkspaceInboundMode = "host_current"; const seedPaths = durableSeedPaths(input.runId); - await ensurePrivateDirectory(descriptorDirectory(input.runId)); + await measureRunPhase("workspace_sync.descriptor_directory", () => ensurePrivateDirectory(descriptorDirectory(input.runId))); try { if (existingReference) { - const existing = await readDescriptor({ + const existing = await measureRunPhase("workspace_sync.descriptor_read", () => readDescriptor({ runId: input.runId, reference: existingReference, - }); + })); if ( existing.descriptor.binding.companyId !== input.companyId || existing.descriptor.binding.workspaceId !== input.workspaceId || @@ -932,12 +937,12 @@ export async function prepareNativeWorkspaceSync(input: { const durableSeed = mode === "durable_seed" ? existing.descriptor.seed - ? await verifiedDurableSeed({ + ? await measureRunPhase("workspace_sync.durable_seed_verify", () => verifiedDurableSeed({ runId: input.runId, - seed: existing.descriptor.seed, + seed: existing.descriptor.seed!, gitSnapshot: existing.descriptor.gitSnapshot, repositories: existing.descriptor.repositories, - }) + })) : (() => { throw new Error("workspace_sync_out_unrecoverable"); })() @@ -1005,16 +1010,16 @@ export async function prepareNativeWorkspaceSync(input: { ); const verifiedWarmAdoption = priorStamp.hostSha256 === currentHostSha256 && - (await remoteStampMatches({ target, expected: priorStamp, gitSnapshot: currentSnapshot.gitSnapshot, repositories: currentSnapshot.repositories })); + (await measureRunPhase("workspace_sync.warm_stamp", () => remoteStampMatches({ target, expected: priorStamp, gitSnapshot: currentSnapshot.gitSnapshot, repositories: currentSnapshot.repositories }))); if (verifiedWarmAdoption) { mode = "adopt_remote"; } else { mode = "host_current"; let preparedSeed: WorkspaceDurableSeedPaths | undefined; try { - preparedSeed = await verifiedDurableSeed({ runId: input.runId, + preparedSeed = await measureRunPhase("workspace_sync.durable_seed_verify", () => verifiedDurableSeed({ runId: input.runId, seed: { workspaceArchiveSha256, gitArchiveSha256 }, - gitSnapshot: currentSnapshot.gitSnapshot, repositories: currentSnapshot.repositories }); + gitSnapshot: currentSnapshot.gitSnapshot, repositories: currentSnapshot.repositories })); } catch (error) { if (!(error instanceof Error) || error.message !== "workspace_sync_out_unrecoverable") throw error; } @@ -1041,7 +1046,7 @@ export async function prepareNativeWorkspaceSync(input: { } const snapshot = runtime.workspaceSyncSnapshot; if (!snapshot) throw new Error("native_workspace_sync_snapshot_missing"); - await persistSnapshotManifests(input.runId, snapshot); + await measureRunPhase("workspace_sync.manifests_persist", () => persistSnapshotManifests(input.runId, snapshot)); const now = new Date().toISOString(); descriptor = { schema: DESCRIPTOR_SCHEMA, @@ -1068,8 +1073,8 @@ export async function prepareNativeWorkspaceSync(input: { } const preparedRuntime = runtime; - let reference = await writeDescriptor(descriptor); - await persistRunReference(input.db, input.runId, reference); + let reference = await measureRunPhase("workspace_sync.descriptor_write", () => writeDescriptor(descriptor)); + await measureRunPhase("workspace_sync.reference_write", () => persistRunReference(input.db, input.runId, reference)); let restorePromise: Promise | null = null; return { mode, diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 3c28f51db62..bb1c0e9192a 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,4 +1,5 @@ import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { measureToolDiscovery, measureRunPhase } from "./execution-preparation-timing.js"; import { composeConnectionInstructions } from "./connection-instructions.js"; import { isInsufficientConnectionScope, INSUFFICIENT_CONNECTION_SCOPE_MESSAGE } from "./connection-permission-errors.js"; import { boundedMcpToolName } from "./mcp-tool-names.js"; @@ -3026,7 +3027,7 @@ export function createToolGatewayService( signal?: AbortSignal, ): Promise { try { - return await toolDiscoveryScheduler.run(() => buildToolsForContext(session, signal), signal); + return await measureToolDiscovery(work => toolDiscoveryScheduler.run(work, signal), () => buildToolsForContext(session, signal)); } catch (error) { if (error instanceof ToolDiscoveryBusyError) { throw new ToolGatewayHttpError(503, error.message, "tool_discovery_busy"); @@ -3037,13 +3038,13 @@ export function createToolGatewayService( /** Shared connection descriptors and task restrictions; no provider calls. */ async function connectionToolsForContext(session: ToolGatewaySession): Promise { - const guestBotConnection = await githubGuestBotConnectionForSession(db, session); - const connectedTools = (await connectedMcpToolsForCompany(session.companyId)) + const guestBotConnection = await measureRunPhase("connectors.discovery.github", () => githubGuestBotConnectionForSession(db, session)); + const connectedTools = (await measureRunPhase("connectors.discovery.connections", () => connectedMcpToolsForCompany(session.companyId))) .filter(tool => !guestBotConnection || !tool.connectionId || (tool.connectionId === guestBotConnection && tool.providerType === "paperclip_github_chat")); return [ - ...await githubBotToolsForSession(db, session), - ...await slackToolsForSession(db, session), - ...await spekoToolsForSession(db, session), + ...await measureRunPhase("connectors.discovery.github", () => githubBotToolsForSession(db, session)), + ...await measureRunPhase("connectors.discovery.slack", () => slackToolsForSession(db, session)), + ...await measureRunPhase("connectors.discovery.speko", () => spekoToolsForSession(db, session)), ...connectedTools, ]; } @@ -3054,7 +3055,7 @@ export function createToolGatewayService( ): Promise { signal?.throwIfAborted(); if (session.agentId) { - await assertAgentInCompany(session.companyId, session.agentId); + await measureRunPhase("connectors.discovery.agent", () => assertAgentInCompany(session.companyId, session.agentId!)); } const connectionTools = await connectionToolsForContext(session); const onDemandTargets = connectionTools.filter(isOnDemandRemoteTool); @@ -3068,12 +3069,12 @@ export function createToolGatewayService( tool.providerType !== "paperclip_plugin"), ); const decisionCache = createToolAccessDecisionCache(); - const decisions = await decideToolsForListing( + const decisions = await measureRunPhase("connectors.discovery.policy", () => decideToolsForListing( tools, (tool) => policyInputForTool({ session, tool }), decisionCache, signal, - ); + )); const visibleTools = decisions .filter( ({ decision }) => @@ -3093,12 +3094,12 @@ export function createToolGatewayService( : tool, ); if (onDemandTargets.length > 0) { - const targetDecisions = await decideToolsForListing( + const targetDecisions = await measureRunPhase("connectors.discovery.on_demand_policy", () => decideToolsForListing( onDemandTargets, (tool) => policyInputForTool({ session, tool }), decisionCache, signal, - ); + )); if ( targetDecisions.some( ({ decision }) => @@ -9134,19 +9135,19 @@ export function createToolGatewayService( async browserUseResources(binding: { companyId: string; agentId: string; runId?: string; issueId?: string }) { if (!binding.runId || !binding.issueId) return []; - const [connection] = await db.select({ id: toolConnections.id }).from(toolConnections).where(and( + const [connection] = await measureRunPhase("connectors.browser_use.connection", () => db.select({ id: toolConnections.id }).from(toolConnections).where(and( eq(toolConnections.companyId, binding.companyId), eq(toolConnections.transport, "rest_api"), eq(toolConnections.enabled, true), eq(toolConnections.status, "active"), sql`${toolConnections.config}->>'sourceTemplateKey' = 'browser-use-cloud'`, - )).limit(1); + )).limit(1)); if (!connection) return []; const session: ToolGatewaySession = { ...binding, runId: binding.runId, issueId: binding.issueId, id: randomUUID(), token: "", projectId: null, createdAt: new Date(), expiresAt: new Date(Date.now() + 60000) }; - const tools = (await listToolsForContext(session)).filter(t => t.providerType === "provider_rest"); + const tools = (await measureRunPhase("connectors.browser_use.catalog", () => listToolsForContext(session))).filter(t => t.providerType === "provider_rest"); const resources: Array<{ id: string; label: string; connectionId: string }> = []; for (const connectionId of new Set(tools.map(t => t.connectionId!))) { - const [connection] = await db.select().from(toolConnections).where(and(eq(toolConnections.companyId, binding.companyId), eq(toolConnections.id, connectionId))); + const [connection] = await measureRunPhase("connectors.browser_use.connection", () => db.select().from(toolConnections).where(and(eq(toolConnections.companyId, binding.companyId), eq(toolConnections.id, connectionId)))); if (!connection || !isBrowserUseConnection(connection)) continue; - try { await resolveConnectionGrant(session, connection, false); resources.push({ id: connectionId, connectionId, label: "Browser Use Cloud" }); } catch { /* Unavailable grants do not contribute a runtime skill. */ } + try { await measureRunPhase("connectors.browser_use.grant", () => resolveConnectionGrant(session, connection, false)); resources.push({ id: connectionId, connectionId, label: "Browser Use Cloud" }); } catch { /* Unavailable grants do not contribute a runtime skill. */ } } return resources; },