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; },