From 6c9b87d0922f1ebfc43e8b0c36889ac5ba7b9837 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:50:51 -0500 Subject: [PATCH] trace(server): expose repository admission waits (#17040) ## Thinking Path > - Paperclip manages work performed by AI agents. > - The controller prepares a task workspace before it starts a provider turn. > - Repository admission also runs when a checkout already exists. > - One observed reuse took 16.877 seconds. Its child spans explained only 3.556 seconds. > - The missing detail covered repeated authority checks, catalog work, and publication waits. > - This change exposes those operations under the existing execution attempt. > - Operators can expand the slow phase before choosing a performance fix. ## Linked Issues or Issue Description Refs #16963, #16330, and #16515. The provider-routing work in #17019 has a different scope. **What existing behavior does this improve?** Repository-admission traces for new and reused checkouts. **Subsystem affected** server/ orchestration and operator-configured OpenTelemetry tracing. **Current behavior** The admission span can contain a long interval without child spans. Existing project checks, HTTP requests, and checkout lookup do not explain current-task checks, database awaits, credential refresh, or publication locks. **Proposed behavior** Expose nested spans for current binding and task access, individual identity reads, catalog construction, credentials, grant transactions, filesystem validation, Git inspection, and directory-lock acquisition, work, and release. **Reason and benefit** The trace can distinguish repeated checks, remote calls, database awaits, and directory contention. A database-await span includes driver, network, pool, and server time. It is not isolated SQL execution time. **Breaking changes** None. No API, schema, permission, retry, lock, or repository behavior changes. ## What Changed - Add fixed admission phases beneath the existing execution attempt. Keep the existing project, checkout, and HTTP detail. - Trace current-task authorization stages only under workspace access. Unrelated decisions cannot consume that detail budget before admission. - Trace GitHub catalog reads, credentials, provider metadata, grant row locking and writes, transaction settlement, and webhook work. - Separate directory-lock acquisition, the protected operation, and release without changing the adapter lock contract. - Document the hierarchy and timing limits. Keep endpoint gating, span caps, and hashed run correlation. Add no span attributes or exception text. ## Verification - Nine new tests passed with real PostgreSQL, real Git, and SDK-exported spans. They cover directory contention, grant-row blocking, all three binding checks, reuse without cloning, dirty files, private-task denial, receipt repair, conflict retry, endpoint-off behavior, and parent IDs. - Twenty earlier unrelated authorization decisions do not exhaust the admission detail budget. - Existing authorization and trust suites: 105 tests passed. Existing timing/context suites: 26 tests passed. Existing repository-access tests: 6 tests passed. - `pnpm -r typecheck` and `pnpm build` passed on commit `5a986d8c86644dfb7ee39b3b832e2d7c86532598`. - `pnpm test:run` could not complete locally. The temporary five-connection cluster returned `sorry, too many clients already` in chat fixtures, and the run was interrupted. It is not a pass. The complete standard CI workflow passed on the same commit, including all 12 server shards and all three chat shards. - Final current-head checks: 54 success, two neutral, two skipped, zero pending or failing. The [CI aggregate](https://github.com/paperclipai/paperclip/actions/runs/39522187940/job/118971261973) passed. [Greptile](https://github.com/paperclipai/paperclip/pull/17040#issuecomment-6305066520) and [Paperclip Review](https://github.com/paperclipai/paperclip/pull/17040#issuecomment-6305017182) both assessed this exact commit at 5/5 with no actionable findings or unresolved threads. - `node scripts/check-module-boundaries.mjs` and `git diff --check` passed. - This macOS host exhausted its SysV semaphore quota. Database tests used a temporary command-scoped initdb preload with five connections, fewer workers, and a larger lock table. No repository test infrastructure or shared host settings changed. Initial skipped tests, focused-fixture failures, and clock-order assertion failures were corrected and rerun; they are not counted as passes. The complete local run remained connection-limited and was interrupted. - Local SDK export proves parentage and privacy. Hosted Honeycomb ingestion remains unverified until deployment. No deployment occurred. ## Risks Span recording requires an operator-configured OTLP endpoint. Each fixed detail name remains capped at 16 occurrences per attempt. Later operations still execute under their enclosing phase. Timing wrappers preserve the existing operations and their errors. No migration is required. The change does not itself reduce admission latency. ## Model Used OpenAI Codex, based on GPT-6, with tool use, code execution, and delegated agents. The exact hosted model identifier 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 | 125 +++++++- .../__tests__/github-catalog-timing.test.ts | 198 ++++++++++++ .../repository-admission-tracing.test.ts | 285 ++++++++++++++++++ server/src/services/authorization.ts | 138 +++++---- .../services/execution-preparation-timing.ts | 30 +- .../services/execution-transaction-timing.ts | 2 +- .../execution-workspace-repositories.ts | 241 +++++++++------ .../services/task-repository-acquisition.ts | 5 +- server/src/services/task-workspace-access.ts | 144 ++++----- server/src/services/tool-access.ts | 118 ++++---- 10 files changed, 989 insertions(+), 297 deletions(-) create mode 100644 server/src/__tests__/github-catalog-timing.test.ts create mode 100644 server/src/__tests__/repository-admission-tracing.test.ts diff --git a/doc/observability.md b/doc/observability.md index 27e8f7e1691..0bc539f500e 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -285,10 +285,11 @@ attempt or owning phase: | Span family | Measured work | | --- | --- | | `workspace.transaction`, `.entry`, `.commit`, `.rollback` | Transaction lifetime, connection acquisition plus BEGIN, and settlement after the callback returns. Entry cannot distinguish pool wait from BEGIN. | -| `workspace.provenance_lock`, `workspace.row_lock` | The awaited advisory-lock statement and current workspace row read with its row lock. | -| `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.row_lock` | Current workspace row read with its row lock. | +| `workspace.sources.inventory`, `workspace.access.*` | Current selected-record reads, task/run identity, current task and selected-project authorization, and selected-source trust. The execution check is separate from full workspace-record disclosure. | +| `workspace.repository.*` | Repository admission binding and policy revalidation, receipt reads/writes, catalog access, filesystem preparation, directory publication lock boundaries, ownership verification, and Git inspection. See the nested admission hierarchy below. | +| `authorization.base`, `.responsible_user`, `.authorizing_user`, `.task_privacy`, `.project_privacy` | The existing base decision, responsible/authorizing-user intersections, and applicable app-record privacy decisions beneath `workspace.access.task_authorization` or `workspace.access.authorization`. These details are scoped to workspace authorization. | +| `workspace.notification`, `workspace.write` | Notification statements and record writes. | | `workspace.materialize.*`, `workspace.checkout.*` | Private source-directory setup and identity verification, existing checkout lookup, Git credential resolution, clone, optional pristine-object cache lookup/publication (`cache_lookup`, `cache_publish`), 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. | @@ -303,6 +304,8 @@ attempt or owning phase: | `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.github.catalog`, `.connections`, `.grants`, `.grant_members`, `.membership`, `.refresh`, `.credentials`, `.repositories`, `.grant_lock`, `.grant_write`, `.webhooks` | Authorized GitHub catalog construction, current connection/grant reads, credential and metadata refresh, repository pagination, grant transaction row lock/write, and parallel webhook binding calls. | +| `connectors.github.refresh.transaction`, `.entry`, `.commit`, `.rollback` | The existing managed-grant refresh transaction and settlement. Entry combines connection acquisition and BEGIN. | | `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. | @@ -312,7 +315,7 @@ attempt or owning phase: | `response.resolve`, `.authorize`, `.add_comment`, `.transaction*`, `.row_lock`, `.write`, `.publish` | Response selection, existing authorization, comment transaction including commit or rollback, row wait, insert, and external publication enqueue. | | `execution.realtime` | Synchronous live-event and plugin publication dispatch. It does not measure browser receipt or paint. | -Detail phases emit at most 16 spans of each name per attempt. Larger reader and +Detail phases emit at most 16 spans of each name per attempt. Larger receipt and asset inventories continue under the containing group without unbounded per-row or per-file spans. Adding detail spans does not itself change work, query order, authorization, or locks. @@ -339,8 +342,116 @@ Existing generations remain under the same storage limits. Cached refs stay private to the company and credential scope. Task checkouts still authenticate online, use the current remote refs, and dissociate their objects from the cache. -Other execution phases carry only the existing hashed run ID; manual boundaries -have no attributes. Exceptions set status without exporting their messages. +Other execution phases, including manual boundaries, carry only the existing +hashed run ID. Exceptions set status without exporting their messages. + +#### Repository admission + +Repository preparation keeps repeated current-identity and permission checks +visible. An existing ready checkout still validates its receipt, origin, pin, +current binding and acquisition authority. A ready checkout can omit clone work +while retaining catalog, database, filesystem and Git inspection work. + +The hierarchy folds those operations beneath the admission that initiated them: + +```text +heartbeat.prepare.workspace_repository_admission + workspace.repository.binding (repeated at publication boundaries) + workspace.repository.db.issue / db.workspace / db.run + workspace.repository.paths (when the destination must be checked) + workspace.repository.current_policy + workspace.repository.db.agent / db.project / db.run + workspace.access.execution + workspace.access.task_read / agent_read / run_read + workspace.access.task_authorization + authorization.base / responsible_user / authorizing_user / task_privacy + workspace.access.workspace_read / source_read / projects_read + workspace.access.authorization (selected project) + authorization.base / responsible_user / authorizing_user / project_privacy + workspace.access.trust + workspace.access.trust.project_read + workspace.repository.receipts + workspace.repository.paths + workspace.repository.catalog + connectors.github.catalog (when GitHub catalog access is needed) + connectors.github.connections / grants / grant_members / membership + connectors.github.refresh (managed grant, including existing retry) + connectors.github.credentials + connectors.github.repositories (metadata and repository pagination) + connectors.github.refresh.transaction + connectors.github.refresh.transaction.entry + connectors.github.grant_lock / grant_write + connectors.github.refresh.transaction.commit / rollback + connectors.github.webhooks (existing parallel binding calls) + connectors.github.credentials / repositories (legacy or unmanaged connection) + workspace.repository.fs_prepare + workspace.repository.ignore_rules + workspace.repository.publication (one per concrete receipt) + workspace.repository.lock_wait + workspace.repository.critical_section + workspace.repository.checkout_identity + workspace.repository.state_write + workspace.repository.ownership_read / ownership_verify + workspace.repository.acquisition (repeated current acquisition check) + workspace.repository.db.issue / db.workspace + workspace.repository.current_policy + workspace.repository.materialize + workspace.checkout.lookup + workspace.checkout.git_auth / git_clone / git_ref / publish (cold path only) + workspace.repository.git_origin / git_branch / git_commit + workspace.repository.binding (same current checks as above) + workspace.repository.ownership_write / state_write + workspace.repository.lock_release +``` + +Optional branches appear only when the corresponding existing operation runs. +The first binding check precedes the receipt lookup; an empty receipt list returns +without catalog, path, staging or publication work. Later binding and acquisition +checks revalidate current authority rather than authorizing historical readers. +The retired filesystem-history inventory, root-policy reservations and reader +authorization phases are no longer part of admission. + +Catalog groups retain the existing sequential connection processing and any +credential refresh or pagination it requires. The initial four catalog reads +can overlap, as can webhook-binding requests. Automatic HTTP spans, when enabled, +sit beneath the owning credential, repository-metadata or webhook group. They +do not require a separate execution-attempt root. `credentials` includes its +existing secret/credential lookup and any OAuth refresh; `repositories` includes +the whole metadata or token-repository listing, rather than one invented span +per repository. No repository list or credential value is attached to a custom +span. + +Database read/write spans measure the complete awaited client operation. Their +duration can include pool acquisition, driver and network time, database work, +and database lock waits. They do not by themselves identify a slow SQL plan or +prove a pool wait. `lock_wait` instead measures the directory publication helper +up to entry into its callback, including canonicalization and lock acquisition. +`critical_section` measures the callback after acquisition. `lock_release` starts +after that callback settles and ends when the helper finishes releasing its +lock. A callback failure marks the critical section and publication as failed; +a successful release remains successful. These directory boundaries are not +PostgreSQL pool or row-lock measurements. + +Parents include their children, so do not add binding, catalog or publication +durations to their descendants. Repeated guards and concurrent HTTP requests +must be read in timeline order. Detail names remain capped at 16 occurrences per +attempt; later operations still execute under their enclosing group. A missing +leaf after that cap, or absent automatic driver instrumentation, does not prove +that no work occurred. + +The `authorization.*` details appear only inside a workspace task/project +authorization group. Unrelated startup decisions do not emit those details or +consume their per-attempt budget before repository admission reaches its current +task checks. + +These additions use the existing opt-in OpenTelemetry path. With +`OTEL_EXPORTER_OTLP_ENDPOINT` unset or blank they perform the original work +without consulting a tracer, even if a provider is globally registered. They add +no Telemetry events, run-log writes or span-attribute keys. The closed attribute +allowlist above remains unchanged: no raw SQL, paths, company/task/workspace or +receipt IDs, repository URLs, refs, credentials, request bodies or error text. +Tracer failures must preserve the original result or error and must not replay +authorization, writes, Git commands or lock callbacks. A committed comment transaction establishes durable server storage. A publication enqueue establishes pending delivery, and realtime dispatch establishes a server diff --git a/server/src/__tests__/github-catalog-timing.test.ts b/server/src/__tests__/github-catalog-timing.test.ts new file mode 100644 index 00000000000..4ee15c6db85 --- /dev/null +++ b/server/src/__tests__/github-catalog-timing.test.ts @@ -0,0 +1,198 @@ +import { randomUUID } from "node:crypto"; +import { context, trace } from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; +import { + authUsers, companies, companyMemberships, connectionGrants, createDb, + EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, toolApplications, toolConnections, +} from "@paperclipai/db"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { writeConnectionCredential } from "../services/connection-credentials.js"; +import { syncConnectionCredentialBindings } from "../services/connection-credential-bindings.js"; +import { withRunExecutionTiming } from "../services/execution-preparation-timing.js"; +import type { PaperclipCloudConnector } from "../services/paperclip-cloud-connector.js"; +import { toolAccessService } from "../services/tool-access.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("GitHub repository catalog execution timing", () => { + let temporary: Awaited>; + let db: ReturnType; + let exporter: InMemorySpanExporter, provider: TracerProvider, manager: AsyncLocalStorageContextManager; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-github-catalog-timing-"); + db = createDb(temporary.connectionString, { maxConnections: 3 }); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + beforeEach(() => { + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "01234567890123456789012345678901"); + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test"); + exporter = new InMemorySpanExporter(); + provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] }); + manager = new AsyncLocalStorageContextManager().enable(); + expect(context.setGlobalContextManager(manager)).toBe(true); + expect(trace.setGlobalTracerProvider(provider)).toBe(true); + }); + afterEach(async () => { + vi.restoreAllMocks(); vi.unstubAllEnvs(); + trace.disable(); context.disable(); manager.disable(); + await provider.shutdown(); + }); + afterAll(async () => { + await db?.$client.end({ timeout: 0 }); + await temporary?.cleanup(); + }); + + async function fixture() { + const userId = `catalog-user-${randomUUID()}`; + const [company] = await db.insert(companies).values({ name: "Catalog timing", issuePrefix: `GH${randomUUID().slice(0, 6)}` }).returning(); + await db.insert(authUsers).values({ id: userId, name: "Catalog owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId: company.id, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" }); + const [application] = await db.insert(toolApplications).values({ companyId: company.id, name: "GitHub catalog", type: "mcp_http" }).returning(); + const [connection] = await db.insert(toolConnections).values({ + companyId: company.id, applicationId: application.id, name: "Private GitHub connection", uid: randomUUID(), + transport: "mcp_remote", authKind: "oauth", credentialPolicy: "per_user", enabled: true, status: "active", + config: { sourceTemplateKey: "github", oauth: { strategy: "paperclip_cloud_connector", connectorProfile: "github.code" } }, + }).returning(); + const token = `private-fixture-token-${randomUUID()}`; + const { secret } = await writeConnectionCredential(db, { + companyId: company.id, connectionName: connection.name, configPath: "oauth.access_token", label: "Access token", + value: token, ownerUserId: userId, actor: { userId }, + }); + const [grant] = await db.insert(connectionGrants).values({ + companyId: company.id, connectionId: connection.id, kind: "user", subjectUserId: userId, + credentialSecretRefs: [{ configPath: "oauth.access_token", secretId: secret.id, versionSelector: "latest" }], + }).returning(); + await syncConnectionCredentialBindings(db, connection); + const webhook = vi.fn(async () => trace.getTracer("catalog-fixture").startActiveSpan("fixture.webhook", async span => { + await Promise.resolve(); span.end(); + })); + const connector = { setWebhookBinding: webhook } as unknown as PaperclipCloudConnector; + const service = toolAccessService(db, { paperclipCloudConnector: connector }); + expect(await service.githubReadHeaders(company.id, connection.id, userId)).toEqual({ Authorization: `Bearer ${token}` }); + const fetchMock = vi.spyOn(globalThis, "fetch").mockImplementation(async url => + trace.getTracer("catalog-fixture").startActiveSpan("fixture.github_http", async span => { + await Promise.resolve(); + span.end(); + const path = new URL(String(url)).pathname; + if (path === "/user") return Response.json({ id: 42, login: "private-fixture-login" }); + if (path === "/user/installations") return Response.json({ installations: [{ id: 101, repository_selection: "selected", account: { login: "private-fixture-org" } }] }); + if (path === "/user/installations/101/repositories") return Response.json({ repositories: [{ id: 201, full_name: "private-fixture-org/repository", private: true }] }); + throw new Error("Unexpected fixture request"); + })); + return { companyId: company.id, userId, connection, grant, service, fetchMock, webhook, token }; + } + + function span(name: string) { + const found = exporter.getFinishedSpans().find(candidate => candidate.name === name); + expect(found, name).toBeDefined(); + return found!; + } + + it("attributes a real grant row-lock wait and preserves provider/write/webhook order", async () => { + const f = await fixture(); + let release!: () => void, acquired!: () => void; + const released = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { acquired = resolve; }); + const blocker = db.transaction(async tx => { + await tx.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)).for("update"); + acquired(); await released; + }); + await ready; + const pending = withRunExecutionTiming("catalog-lock", () => f.service.listGitHubRepositories(f.companyId, f.userId)); + void pending.catch(() => undefined); + try { + await vi.waitFor(() => { expect(f.fetchMock).toHaveBeenCalledTimes(3); }); + await vi.waitFor(async () => { + const waiting = await db.execute(sql`select count(*)::int as count from pg_locks where not granted and locktype = ${"transactionid"}`); + expect(waiting[0]?.count).toBeGreaterThan(0); + }); + expect(exporter.getFinishedSpans().some(candidate => candidate.name === "connectors.github.grant_lock")).toBe(false); + expect(f.webhook).not.toHaveBeenCalled(); + } finally { + release(); await blocker; await pending; + } + const result = await pending; + expect(result).toMatchObject({ connectionCount: 1, failedConnectionCount: 0, repositories: [{ id: "201", fullName: "private-fixture-org/repository" }] }); + const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.grant.id)); + expect(saved.providerTenant?.github?.repositoryCount).toBe(1); + expect(f.webhook).toHaveBeenCalledTimes(1); + const catalog = span("connectors.github.catalog"); + const refresh = span("connectors.github.refresh"); + const transaction = span("connectors.github.refresh.transaction"); + expect(catalog.parentSpanContext?.spanId).toBe(span("task.execution_attempt").spanContext().spanId); + expect(refresh.parentSpanContext?.spanId).toBe(catalog.spanContext().spanId); + expect(transaction.parentSpanContext?.spanId).toBe(refresh.spanContext().spanId); + for (const name of ["grant_lock", "grant_write"]) expect(span(`connectors.github.${name}`).parentSpanContext?.spanId).toBe(transaction.spanContext().spanId); + for (const name of ["entry", "commit"]) expect(span(`connectors.github.refresh.transaction.${name}`).parentSpanContext?.spanId).toBe(transaction.spanContext().spanId); + expect(span("fixture.github_http").parentSpanContext?.spanId).toBe(span("connectors.github.repositories").spanContext().spanId); + expect(span("fixture.webhook").parentSpanContext?.spanId).toBe(span("connectors.github.webhooks").spanContext().spanId); + for (const name of ["connections", "grants", "grant_members", "membership"]) expect(span(`connectors.github.${name}`).parentSpanContext?.spanId).toBe(catalog.spanContext().spanId); + const exported = JSON.stringify(exporter.getFinishedSpans().map(({ name, attributes, events }) => ({ name, attributes, events }))); + for (const privateValue of [f.companyId, f.userId, f.connection.id, f.connection.name, f.grant.id, f.token, "private-fixture-org", "private-fixture-login"]) expect(exported).not.toContain(privateValue); + }); + + it("preserves denied audiences and swallowed provider failures without exporting error text", async () => { + const f = await fixture(); + const denied = await withRunExecutionTiming("catalog-denied", () => f.service.listGitHubRepositories(f.companyId, "unrelated-user")); + expect(denied).toMatchObject({ repositories: [], connectionCount: 0, failedConnectionCount: 0 }); + expect(f.fetchMock).not.toHaveBeenCalled(); + const failure = new Error(`private-provider-error-${randomUUID()}`); + f.fetchMock.mockRejectedValue(failure); + const result = await withRunExecutionTiming("catalog-failed", () => f.service.listGitHubRepositories(f.companyId, f.userId)); + expect(result).toMatchObject({ repositories: [], connectionCount: 1, failedConnectionCount: 1 }); + expect(span("connectors.github.repositories").status.code).toBe(2); + expect(span("connectors.github.refresh").status.code).toBe(2); + expect(JSON.stringify(exporter.getFinishedSpans().map(({ attributes, events }) => ({ attributes, events })))).not.toContain(failure.message); + expect(f.webhook).not.toHaveBeenCalled(); + }); + + it("retains the single access-conflict retry and records rollback before the successful commit", async () => { + const f = await fixture(); + const request = f.fetchMock.getMockImplementation()!; + let changed = false; + f.fetchMock.mockImplementation(async (...args) => { + const response = await request(...args); + if (!changed && new URL(String(args[0])).pathname.endsWith("/repositories")) { + changed = true; + await db.update(connectionGrants).set({ providerTenant: { github: { + userId: "42", login: "private-fixture-login", installationCount: 0, repositoryCount: 0, + repositorySelection: "none", installationIds: [], installationOwnerLogins: [], accessRevision: "concurrent-webhook", + } } }).where(eq(connectionGrants.id, f.grant.id)); + } + return response; + }); + const result = await withRunExecutionTiming("catalog-conflict", () => f.service.listGitHubRepositories(f.companyId, f.userId)); + expect(result).toMatchObject({ connectionCount: 1, failedConnectionCount: 0, repositories: [{ id: "201" }] }); + expect(f.fetchMock).toHaveBeenCalledTimes(6); + expect(f.webhook).toHaveBeenCalledTimes(1); + const transactions = exporter.getFinishedSpans().filter(candidate => candidate.name === "connectors.github.refresh.transaction"); + expect(transactions).toHaveLength(2); + expect(transactions.map(candidate => candidate.status.code)).toEqual([2, 0]); + const rollback = span("connectors.github.refresh.transaction.rollback"); + const commit = span("connectors.github.refresh.transaction.commit"); + expect(rollback.parentSpanContext?.spanId).toBe(transactions[0].spanContext().spanId); + expect(commit.parentSpanContext?.spanId).toBe(transactions[1].spanContext().spanId); + const retryRead = exporter.getFinishedSpans().find(candidate => candidate.name === "connectors.github.grants" + && candidate.parentSpanContext?.spanId === span("connectors.github.refresh").spanContext().spanId); + expect(retryRead).toBeDefined(); + }); + + it("keeps the same catalog outputs with endpoint disabled and emits no owned-run spans outside an attempt", async () => { + const f = await fixture(); + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", ""); + const disabled = await withRunExecutionTiming("catalog-off", () => f.service.listGitHubRepositories(f.companyId, f.userId)); + expect(disabled).toMatchObject({ connectionCount: 1, failedConnectionCount: 0, repositories: [{ id: "201" }] }); + expect(exporter.getFinishedSpans().filter(candidate => candidate.name.startsWith("connectors.github."))).toEqual([]); + exporter.reset(); + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test"); + const outside = await f.service.listGitHubRepositories(f.companyId, f.userId); + expect(outside).toEqual(disabled); + expect(exporter.getFinishedSpans().filter(candidate => candidate.name.startsWith("connectors.github."))).toEqual([]); + exporter.reset(); + const enabled = await withRunExecutionTiming("catalog-on", () => f.service.listGitHubRepositories(f.companyId, f.userId)); + expect(enabled).toEqual(disabled); + expect(span("connectors.github.catalog").status.code).toBe(0); + }); +}); diff --git a/server/src/__tests__/repository-admission-tracing.test.ts b/server/src/__tests__/repository-admission-tracing.test.ts new file mode 100644 index 00000000000..0ca0f9c17b6 --- /dev/null +++ b/server/src/__tests__/repository-admission-tracing.test.ts @@ -0,0 +1,285 @@ +import { execFile } from "node:child_process"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import { context, trace } from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider, type ReadableSpan, type SpanProcessor } from "@opentelemetry/sdk-trace"; +import { withDirectoryPublicationLock } from "@paperclipai/adapter-utils/workspace-restore-merge"; +import { createDb, executionWorkspaceRepositories, executionWorkspaces, issues, projects, projectWorkspaces, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createRunPreparationTiming, withRunExecutionTiming } from "../services/execution-preparation-timing.js"; +import { executionWorkspaceRepositoryService } from "../services/execution-workspace-repositories.js"; +import { authorizationService } from "../services/authorization.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { addReadyWorkspaceHistoryRepository, createWorkspacePreparationHistoryFixture } from "./helpers/workspace-preparation-history-fixture.js"; + +// Only the external catalog is replaced. Identity, authorization, receipts, +// publication locks and existing-checkout inspection use their real implementations. +const catalog = vi.hoisted(() => ({ list: vi.fn(async () => ({ repositories: [] })) })); +vi.mock("../services/tool-access.js", () => ({ toolAccessService: () => ({ listProjectRepositories: catalog.list }) })); + +const exec = promisify(execFile); +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("repository admission tracing with real persistence and Git", () => { + let database: Awaited>; + let db: ReturnType; + let directory: string; + let fixture: Awaited>; + let exporter: InMemorySpanExporter; + let provider: TracerProvider; + let manager: AsyncLocalStorageContextManager; + let started: string[]; + let lifecycle: string[]; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-repository-admission-tracing-"); + db = createDb(database.connectionString); + directory = await fs.mkdtemp(path.join(os.tmpdir(), "repository-admission-tracing-")); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + + beforeEach(async () => { + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test"); + vi.stubEnv("PAPERCLIP_ISSUE_PRIVACY_MODE", "enforce"); + vi.stubEnv("PAPERCLIP_HOME", path.join(directory, "home")); + vi.stubEnv("PAPERCLIP_INSTANCE_ID", "admission-tracing"); + catalog.list.mockClear(); + started = []; + lifecycle = []; + exporter = new InMemorySpanExporter(); + const starts: SpanProcessor = { + onStart: span => { started.push(span.name); lifecycle.push(`${span.spanContext().spanId}:start`); }, + onEnd: span => { lifecycle.push(`${span.spanContext().spanId}:end`); }, + forceFlush: async () => {}, shutdown: async () => {}, + }; + provider = new TracerProvider({ spanProcessors: [starts, new SimpleSpanProcessor({ exporter })] }); + manager = new AsyncLocalStorageContextManager().enable(); + expect(context.setGlobalContextManager(manager)).toBe(true); + expect(trace.setGlobalTracerProvider(provider)).toBe(true); + fixture = await createWorkspacePreparationHistoryFixture(db, path.join(directory, `source-${Date.now()}`), { + workspaces: 0, sharedRoots: 0, sources: 0, readers: 0, + }); + const [project] = await db.insert(projects).values({ companyId: fixture.companyId, name: "Selected source" }).returning(); + const [source] = await db.insert(projectWorkspaces).values({ companyId: fixture.companyId, + projectId: project.id, name: "Selected checkout", cwd: fixture.cwd }).returning(); + await db.update(issues).set({ projectWorkspaceId: source.id }).where(eq(issues.id, fixture.issueId)); + await db.update(executionWorkspaces).set({ projectId: project.id, projectWorkspaceId: source.id }) + .where(eq(executionWorkspaces.id, fixture.workspaceId)); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + trace.disable(); context.disable(); manager.disable(); + await provider.shutdown(); + }); + + afterAll(async () => { + await db?.$client.end({ timeout: 0 }); + await database?.cleanup(); + if (directory) await fs.rm(directory, { recursive: true, force: true }); + }); + + function admit(beforeAdmission?: () => Promise) { + return withRunExecutionTiming(fixture.runId, async () => { + await beforeAdmission?.(); + return createRunPreparationTiming(fixture.runId)("workspace_repository_admission", + () => executionWorkspaceRepositoryService(db).prepareForAdmission(fixture.input)); + }); + } + + function only(spans: ReadableSpan[], name: string, parent?: ReadableSpan) { + const found = spans.filter(span => span.name === name + && (!parent || span.parentSpanContext?.spanId === parent.spanContext().spanId)); + expect(found, `${name} under ${parent?.name ?? "trace"}`).toHaveLength(1); + return found[0]!; + } + + // SDK spans anchor their wall clocks independently at millisecond precision. + // Processor event order proves causal containment without sub-millisecond jitter. + const startsAt = (span: ReadableSpan) => lifecycle.indexOf(`${span.spanContext().spanId}:start`); + const endsAt = (span: ReadableSpan) => lifecycle.indexOf(`${span.spanContext().spanId}:end`); + + function expectChild(child: ReadableSpan, parent: ReadableSpan) { + expect(child.parentSpanContext?.spanId).toBe(parent.spanContext().spanId); + expect(child.spanContext().traceId).toBe(parent.spanContext().traceId); + expect(startsAt(child)).toBeGreaterThanOrEqual(startsAt(parent)); + expect(endsAt(child)).toBeLessThanOrEqual(endsAt(parent)); + } + + async function recordGitCommands() { + const realGit = (await exec("which", ["git"])).stdout.trim(); + const bin = path.join(fixture.cwd, "test-bin"); + const log = path.join(fixture.cwd, "git-commands.jsonl"); + await fs.mkdir(bin); + // Pass through to real Git and retain only local synthetic test evidence. + await fs.writeFile(path.join(bin, "git"), `#!${process.execPath}\nconst fs = require("node:fs");\nconst cp = require("node:child_process");\nconst args = process.argv.slice(2);\nfs.appendFileSync(${JSON.stringify(log)}, JSON.stringify(args) + "\\n");\nconst result = cp.spawnSync(${JSON.stringify(realGit)}, args, {stdio: "inherit", env: process.env});\nprocess.exit(result.status ?? 1);\n`, { mode: 0o755 }); + vi.stubEnv("PATH", `${bin}${path.delimiter}${process.env.PATH}`); + return async () => (await fs.readFile(log, "utf8")).trim().split("\n").map(line => JSON.parse(line) as string[]); + } + + function expectPrivateFieldsAbsent(spans: ReadableSpan[]) { + for (const span of spans) { + expect(Object.keys(span.attributes)).toEqual(["paperclip.task.run.run_id"]); + expect(span.events).toEqual([]); + expect(span.status.message).toBeUndefined(); + } + const exported = JSON.stringify(spans.map(span => ({ attributes: span.attributes, events: span.events, status: span.status }))); + for (const privateValue of [fixture.companyId, fixture.issueId, fixture.workspaceId, fixture.runId, fixture.cwd, fixture.pin]) { + expect(exported).not.toContain(privateValue); + } + } + + it("exports repeated current checks and separates real lock contention from ready-checkout work", async () => { + const repository = await addReadyWorkspaceHistoryRepository(db, fixture); + const dirtyFile = path.join(repository.checkout, "result.txt"); + await fs.appendFile(dirtyFile, "Uncommitted task work\n"); + const commands = await recordGitCommands(); + let release!: () => void, acquired!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { acquired = resolve; }); + const blocker = withDirectoryPublicationLock(repository.checkout, async () => { acquired(); await gate; }); + await ready; + const admission = admit(async () => { + const authorization = authorizationService(db); + // Startup performs many unrelated decisions before repository admission. + // They must not exhaust the workspace authorization detail budget. + for (let index = 0; index < 20; index++) { + const decision = await authorization.decide({ actor: { type: "board", source: "local_implicit", userId: "local-board" }, + action: "company_scope:read", resource: { type: "company", companyId: fixture.companyId } }); + expect(decision.allowed).toBe(true); + } + }); + void admission.catch(() => undefined); + try { + await vi.waitFor(() => { + expect(started).toContain("workspace.repository.lock_wait"); + expect(exporter.getFinishedSpans().some(span => span.name === "workspace.repository.lock_wait")).toBe(false); + expect(started).not.toContain("workspace.repository.critical_section"); + }, { timeout: 10_000 }); + } finally { + release(); + await blocker; + await admission; + } + await expect(admission).resolves.toEqual([{ id: repository.row.id, cwd: repository.checkout, + repoUrl: repository.row.repoUrl, relativePath: repository.row.relativePath, pinnedCommit: fixture.pin, branchName: "main" }]); + expect(catalog.list).toHaveBeenCalledOnce(); + expect(await fs.readFile(dirtyFile, "utf8")).toBe("Synthetic task workspace\nUncommitted task work\n"); + expect((await commands()).some(args => args.includes("clone"))).toBe(false); + const [saved] = await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, repository.row.id)); + expect(saved).toMatchObject({ state: "ready", pinnedCommit: fixture.pin, branchName: "main", failureCode: null }); + + const spans = exporter.getFinishedSpans(); + const attempt = only(spans, "task.execution_attempt"); + const phase = only(spans, "heartbeat.prepare.workspace_repository_admission", attempt); + for (const span of spans.filter(span => span !== attempt)) { + const parent = spans.find(candidate => candidate.spanContext().spanId === span.parentSpanContext?.spanId); + expect(parent, `parent of ${span.name}`).toBeDefined(); + expectChild(span, parent!); + if (span.name.startsWith("authorization.")) { + expect(["workspace.access.task_authorization", "workspace.access.authorization"]).toContain(parent!.name); + } + } + const bindings = spans.filter(span => span.name === "workspace.repository.binding"); + expect(bindings).toHaveLength(3); + for (const binding of bindings) { + const execution = only(spans, "workspace.access.execution", binding); + for (const name of ["task_read", "agent_read", "run_read", "workspace_read", "source_read", "projects_read", "trust"]) { + expectChild(only(spans, `workspace.access.${name}`, execution), execution); + } + const taskAuthorization = only(spans, "workspace.access.task_authorization", execution); + for (const name of ["base", "responsible_user", "authorizing_user", "task_privacy"]) { + expectChild(only(spans, `authorization.${name}`, taskAuthorization), taskAuthorization); + } + for (const name of ["issue", "workspace", "run"]) expectChild(only(spans, `workspace.repository.db.${name}`, binding), binding); + expectChild(only(spans, "workspace.repository.current_policy", binding), binding); + } + const publication = only(spans, "workspace.repository.publication", phase); + const wait = only(spans, "workspace.repository.lock_wait", publication); + const critical = only(spans, "workspace.repository.critical_section", publication); + const releaseSpan = only(spans, "workspace.repository.lock_release", publication); + expect(endsAt(wait)).toBeLessThanOrEqual(startsAt(critical)); + expect(endsAt(critical)).toBeLessThanOrEqual(startsAt(releaseSpan)); + expectChild(releaseSpan, publication); + for (const name of ["checkout_identity", "ownership_verify", "materialize", "git_origin", "git_branch"]) { + expectChild(only(spans, `workspace.repository.${name}`, critical), critical); + } + expectChild(only(spans, "workspace.checkout.lookup"), only(spans, "workspace.repository.materialize", critical)); + expect(spans.filter(span => span.name === "workspace.repository.state_write")).toHaveLength(2); + expect(spans.filter(span => span.name === "workspace.repository.acquisition")).toHaveLength(2); + expect(spans.some(span => span.name === "workspace.checkout.git_clone")).toBe(false); + expect(spans.every(span => span.status.code !== 2)).toBe(true); + expectPrivateFieldsAbsent(spans); + }); + + it("keeps ready admission and its checkout unchanged when the endpoint is disabled", async () => { + const repository = await addReadyWorkspaceHistoryRepository(db, fixture); + const commands = await recordGitCommands(); + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", ""); + const tracer = vi.spyOn(trace, "getTracer"); + await expect(admit()).resolves.toMatchObject([{ id: repository.row.id, pinnedCommit: fixture.pin, cwd: repository.checkout }]); + expect(tracer).not.toHaveBeenCalled(); + expect(exporter.getFinishedSpans()).toEqual([]); + expect(catalog.list).toHaveBeenCalledOnce(); + expect((await commands()).some(args => args.includes("clone"))).toBe(false); + expect(await fs.readFile(path.join(repository.checkout, "result.txt"), "utf8")).toBe("Synthetic task workspace\n"); + }); + + it("preserves receipt repair and releases the publication lock after a real integrity failure", async () => { + const repository = await addReadyWorkspaceHistoryRepository(db, fixture); + const markerPath = path.join(repository.checkout, ".git", "paperclip-workspace-owner.json"); + const invalidMarker = JSON.stringify({ version: 1, repositoryId: repository.row.id, pinnedCommit: "1".repeat(40) }); + await fs.writeFile(markerPath, invalidMarker); + await expect(admit()).rejects.toMatchObject({ status: 409, + message: "Shared repository publication receipts disagree; repair is required before reuse" }); + expect(catalog.list).toHaveBeenCalledOnce(); + expect(await fs.readFile(markerPath, "utf8")).toBe(invalidMarker); + const [saved] = await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, repository.row.id)); + expect(saved).toMatchObject({ state: "needs_repair", failureCode: "repository_preparation_failed", pinnedCommit: fixture.pin }); + const afterRelease = vi.fn(async () => undefined); + await withDirectoryPublicationLock(repository.checkout, afterRelease); + expect(afterRelease).toHaveBeenCalledOnce(); + const spans = exporter.getFinishedSpans(); + const publication = only(spans, "workspace.repository.publication"); + const critical = only(spans, "workspace.repository.critical_section", publication); + const release = only(spans, "workspace.repository.lock_release", publication); + expect(critical.status.code).toBe(2); + expect(publication.status.code).toBe(2); + expect(only(spans, "workspace.repository.ownership_verify", critical).status.code).toBe(2); + expect(release.status.code).not.toBe(2); + expectChild(release, publication); + expect(endsAt(critical)).toBeLessThanOrEqual(startsAt(release)); + expect(spans.filter(span => span.name === "workspace.repository.state_write")).toHaveLength(2); + expect(spans.some(span => span.name === "workspace.repository.materialize")).toBe(false); + expectPrivateFieldsAbsent(spans); + }); + + it("exports the current task authorization denial before any repository or filesystem work", async () => { + await db.update(issues).set({ visibility: "private", privacyRootIssueId: fixture.issueId, + responsibleUserId: fixture.principals[1].userId }).where(eq(issues.id, fixture.issueId)); + await expect(admit()).rejects.toMatchObject({ status: 403 }); + expect(catalog.list).not.toHaveBeenCalled(); + const spans = exporter.getFinishedSpans(); + const execution = only(spans, "workspace.access.execution"); + const authorization = only(spans, "workspace.access.task_authorization", execution); + expectChild(only(spans, "authorization.task_privacy", authorization), authorization); + expect(execution.status.code).toBe(2); + expect(only(spans, "workspace.repository.binding").status.code).toBe(2); + expect(spans.some(span => ["workspace.repository.receipts", "workspace.repository.paths", "workspace.repository.publication"].includes(span.name))).toBe(false); + expectPrivateFieldsAbsent(spans); + }); + + it("returns empty receipts after current checks without catalog, path checks or publication", async () => { + await expect(admit()).resolves.toEqual([]); + expect(catalog.list).not.toHaveBeenCalled(); + const spans = exporter.getFinishedSpans(); + const phase = only(spans, "heartbeat.prepare.workspace_repository_admission"); + only(spans, "workspace.repository.binding", phase); + only(spans, "workspace.repository.receipts", phase); + expect(spans.some(span => ["workspace.repository.paths", "workspace.repository.fs_prepare", "workspace.repository.publication"].includes(span.name))).toBe(false); + expectPrivateFieldsAbsent(spans); + }); +}); diff --git a/server/src/services/authorization.ts b/server/src/services/authorization.ts index 539497760e6..71d4affa174 100644 --- a/server/src/services/authorization.ts +++ b/server/src/services/authorization.ts @@ -48,6 +48,12 @@ import { import { logger } from "../middleware/logger.js"; import { normalizeAgentPermissions } from "../lib/agent-permissions.js"; import { grantsForHumanRole, normalizeHumanRole } from "./company-member-roles.js"; +import { isRunPhaseActive, measureRunPhase, type ExecutionPhase } from "./execution-preparation-timing.js"; + +function measureWorkspaceAuthorization(phase: Extract, work: () => T | Promise): T | Promise { + return isRunPhaseActive("workspace.access.task_authorization") || isRunPhaseActive("workspace.access.authorization") + ? measureRunPhase(phase, work) : work(); +} export type AuthorizationActor = { @@ -2789,9 +2795,9 @@ function createAuthorizationService(db: Db | DbTransaction, readSnapshot?: ReadA resource: AuthorizationResource; scope?: Record | null; }): Promise { - const agentDecision = await decideBase(input); - const responsibleDecision = await applyResponsibleUserIntersection(input, agentDecision); - const intersectedDecision = await applyAuthorizingUserIntersection(input, responsibleDecision); + const agentDecision = await measureWorkspaceAuthorization("authorization.base", () => decideBase(input)); + const responsibleDecision = await measureWorkspaceAuthorization("authorization.responsible_user", () => applyResponsibleUserIntersection(input, agentDecision)); + const intersectedDecision = await measureWorkspaceAuthorization("authorization.authorizing_user", () => applyAuthorizingUserIntersection(input, responsibleDecision)); if ( input.action === "project:read" && input.resource.type === "project" @@ -2801,31 +2807,36 @@ function createAuthorizationService(db: Db | DbTransaction, readSnapshot?: ReadA && intersectedDecision.reason !== "allow_local_board" && (input.actor.authorizingUserId || issuePrivacyMode() !== "off") ) { - const retainedProject = readSnapshot?.projects.get(input.resource.projectId); - const project = retainedProject?.companyId === input.resource.companyId ? retainedProject : await db - .select({ id: projects.id, companyId: projects.companyId, visibility: projects.visibility }) - .from(projects) - .where(and(eq(projects.id, input.resource.projectId), eq(projects.companyId, input.resource.companyId))) - .then((rows) => rows[0] ?? null); - if (project && !(retainedProject?.companyId === input.resource.companyId - ? retainedProject.readable : await canActorReadProjectPrivacy(db, input.actor, project))) { - const denied = deny({ - action: input.action, - reason: "deny_project_private", - explanation: "Project is private and the actor is not an access member.", - }); - logger.warn({ - authzMode: issuePrivacyMode(), - action: input.action, - companyId: project.companyId, - projectId: project.id, - actorType: input.actor.type, - actorUserId: input.actor.type === "board" ? input.actor.userId ?? null : null, - actorAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, - reason: denied.reason, - }, "project privacy would deny read"); - return !input.actor.authorizingUserId && issuePrivacyMode() === "shadow" ? intersectedDecision : denied; - } + const resource = input.resource; + const privacyDecision = await measureWorkspaceAuthorization("authorization.project_privacy", async () => { + const retainedProject = readSnapshot?.projects.get(resource.projectId!); + const project = retainedProject?.companyId === resource.companyId ? retainedProject : await db + .select({ id: projects.id, companyId: projects.companyId, visibility: projects.visibility }) + .from(projects) + .where(and(eq(projects.id, resource.projectId!), eq(projects.companyId, resource.companyId))) + .then((rows) => rows[0] ?? null); + if (project && !(retainedProject?.companyId === resource.companyId + ? retainedProject.readable : await canActorReadProjectPrivacy(db, input.actor, project))) { + const denied = deny({ + action: input.action, + reason: "deny_project_private", + explanation: "Project is private and the actor is not an access member.", + }); + logger.warn({ + authzMode: issuePrivacyMode(), + action: input.action, + companyId: project.companyId, + projectId: project.id, + actorType: input.actor.type, + actorUserId: input.actor.type === "board" ? input.actor.userId ?? null : null, + actorAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, + reason: denied.reason, + }, "project privacy would deny read"); + return !input.actor.authorizingUserId && issuePrivacyMode() === "shadow" ? intersectedDecision : denied; + } + return null; + }); + if (privacyDecision) return privacyDecision; } if ( input.action !== "issue:read" @@ -2837,42 +2848,45 @@ function createAuthorizationService(db: Db | DbTransaction, readSnapshot?: ReadA || (!input.actor.authorizingUserId && issuePrivacyMode() === "off") ) return intersectedDecision; - const retainedIssue = readSnapshot?.issues.get(input.resource.issueId); - const issue = retainedIssue?.companyId === input.resource.companyId ? retainedIssue : await db - .select({ - id: issues.id, - companyId: issues.companyId, - visibility: issues.visibility, - privacyRootIssueId: issues.privacyRootIssueId, - responsibleUserId: issues.responsibleUserId, - createdByUserId: issues.createdByUserId, - assigneeUserId: issues.assigneeUserId, - assigneeAgentId: issues.assigneeAgentId, - projectId: issues.projectId, - }) - .from(issues) - .where(and(eq(issues.id, input.resource.issueId), eq(issues.companyId, input.resource.companyId))) - .then((rows) => rows[0] ?? null); - if (!issue || (retainedIssue?.companyId === input.resource.companyId - ? retainedIssue.readable : await canActorReadIssuePrivacy(db, input.actor, issue))) return intersectedDecision; + const resource = input.resource; + return measureWorkspaceAuthorization("authorization.task_privacy", async () => { + const retainedIssue = readSnapshot?.issues.get(resource.issueId!); + const issue = retainedIssue?.companyId === resource.companyId ? retainedIssue : await db + .select({ + id: issues.id, + companyId: issues.companyId, + visibility: issues.visibility, + privacyRootIssueId: issues.privacyRootIssueId, + responsibleUserId: issues.responsibleUserId, + createdByUserId: issues.createdByUserId, + assigneeUserId: issues.assigneeUserId, + assigneeAgentId: issues.assigneeAgentId, + projectId: issues.projectId, + }) + .from(issues) + .where(and(eq(issues.id, resource.issueId!), eq(issues.companyId, resource.companyId))) + .then((rows) => rows[0] ?? null); + if (!issue || (retainedIssue?.companyId === resource.companyId + ? retainedIssue.readable : await canActorReadIssuePrivacy(db, input.actor, issue))) return intersectedDecision; - const denied = deny({ - action: input.action, - reason: "deny_issue_private", - explanation: "Issue is private and the actor is not an implicit principal or active grantee.", + const denied = deny({ + action: input.action, + reason: "deny_issue_private", + explanation: "Issue is private and the actor is not an implicit principal or active grantee.", + }); + logger.warn({ + authzMode: issuePrivacyMode(), + action: input.action, + companyId: issue.companyId, + issueId: issue.id, + privacyRootIssueId: issue.privacyRootIssueId, + actorType: input.actor.type, + actorUserId: input.actor.type === "board" ? input.actor.userId ?? null : null, + actorAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, + reason: denied.reason, + }, "issue privacy would deny read"); + return !input.actor.authorizingUserId && issuePrivacyMode() === "shadow" ? intersectedDecision : denied; }); - logger.warn({ - authzMode: issuePrivacyMode(), - action: input.action, - companyId: issue.companyId, - issueId: issue.id, - privacyRootIssueId: issue.privacyRootIssueId, - actorType: input.actor.type, - actorUserId: input.actor.type === "board" ? input.actor.userId ?? null : null, - actorAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, - reason: denied.reason, - }, "issue privacy would deny read"); - return !input.actor.authorizingUserId && issuePrivacyMode() === "shadow" ? intersectedDecision : denied; } // A candidate filter only: project policies and responsible-user grants are diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index e92e42ee985..214838d5d38 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -40,6 +40,26 @@ 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.access.execution" | "workspace.access.task_read" | "workspace.access.agent_read" | "workspace.access.run_read" + | "workspace.access.workspace_read" | "workspace.access.source_read" | "workspace.access.projects_read" + | "workspace.access.task_authorization" | "workspace.access.trust" | "workspace.access.trust.project_read" + | "authorization.base" | "authorization.responsible_user" | "authorization.authorizing_user" + | "authorization.task_privacy" | "authorization.project_privacy" + | "workspace.repository.binding" | "workspace.repository.current_policy" | "workspace.repository.acquisition" + | "workspace.repository.catalog" | "workspace.repository.receipts" | "workspace.repository.paths" + | "workspace.repository.fs_prepare" | "workspace.repository.publication" | "workspace.repository.lock_wait" + | "workspace.repository.critical_section" | "workspace.repository.lock_release" | "workspace.repository.checkout_identity" + | "workspace.repository.state_write" | "workspace.repository.ownership_read" | "workspace.repository.ownership_verify" + | "workspace.repository.materialize" | "workspace.repository.git_origin" | "workspace.repository.git_branch" + | "workspace.repository.git_commit" | "workspace.repository.ownership_write" | "workspace.repository.ignore_rules" + | "workspace.repository.db.issue" | "workspace.repository.db.workspace" | "workspace.repository.db.agent" + | "workspace.repository.db.project" | "workspace.repository.db.run" + | "connectors.github.catalog" | "connectors.github.connections" | "connectors.github.grants" + | "connectors.github.grant_members" | "connectors.github.membership" + | "connectors.github.refresh" | "connectors.github.credentials" | "connectors.github.repositories" + | "connectors.github.grant_lock" | "connectors.github.grant_write" | "connectors.github.webhooks" + | "connectors.github.refresh.transaction" | "connectors.github.refresh.transaction.entry" + | "connectors.github.refresh.transaction.commit" | "connectors.github.refresh.transaction.rollback" | "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" @@ -80,7 +100,7 @@ export type ExecutionPhase = | "response.transaction" | "response.transaction.entry" | "response.transaction.commit" | "response.transaction.rollback" | "response.row_lock" | "response.write" | "response.resolve" | "response.authorize" | "response.add_comment" | "response.activity" | "response.publish"; -type RunScope = { runId: string; parent?: Context; ended: boolean; details: Map }; +type RunScope = { runId: string; phase: string; parent?: Context; ended: boolean; details: Map }; const ownership = new AsyncLocalStorage(); const endpointEnabled = () => Boolean(process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim()); @@ -109,12 +129,18 @@ export function getRunExecutionTraceContext(runId: string): Context | undefined return endpointEnabled() && owned?.runId === runId && !owned.ended ? owned.parent : undefined; } +/** Inspect only the current async scope; unrelated authorization must not consume its detail budget. */ +export function isRunPhaseActive(phase: ExecutionPhase): boolean { + const owned = ownership.getStore(); + return endpointEnabled() && owned?.phase === phase && !owned.ended; +} + function measure(runId: string, name: string, work: () => T | Promise, independent = false, counts: { fileCount?: number; bytes?: number } = {}): Promise { let operation: Promise | undefined; const run = (span?: { setStatus(status: { code: number }): void; end(): void }, parent?: Context) => { if (operation) return operation; const previous = ownership.getStore(); - const owned: RunScope = { runId, + const owned: RunScope = { runId, phase: name, parent: parent ?? (!independent && previous?.runId === runId && !previous.ended ? previous.parent : undefined), ended: false, details: !independent && previous?.runId === runId ? previous.details : new Map(), diff --git a/server/src/services/execution-transaction-timing.ts b/server/src/services/execution-transaction-timing.ts index 1cf4583fcd9..77b3bd038f5 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" | "connectors.slack.authority" | "connectors.slack.catalog", work: (tx: Transaction) => Promise): Promise { +export function measureRunTransaction(db: Db, scope: "workspace" | "runtime_selection" | "response" | "connectors.slack.authority" | "connectors.slack.catalog" | "connectors.github.refresh", 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/execution-workspace-repositories.ts b/server/src/services/execution-workspace-repositories.ts index 665630edbcc..f0eec1af1b3 100644 --- a/server/src/services/execution-workspace-repositories.ts +++ b/server/src/services/execution-workspace-repositories.ts @@ -1,5 +1,5 @@ import { measureRunTransaction } from "./execution-transaction-timing.js"; -import { measureRunPhase } from "./execution-preparation-timing.js"; +import { beginRunPhase, measureRunPhase } from "./execution-preparation-timing.js"; import { createHash, randomUUID } from "node:crypto"; import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; @@ -31,6 +31,40 @@ type DbTransaction = Parameters[0]>[0]; const controllerAcquisitionMessage = "New controller repository acquisition requires a trusted task. Existing verified public checkouts remain available."; const lowTrustRepositoryMessage = "Low-trust repository reuse permits anonymous public URLs only; new acquisition is unavailable. Use an authorized configured source for private repositories."; +function withRepositoryPublication(cwd: string, work: () => Promise): Promise { + return measureRunPhase("workspace.repository.publication", async () => { + const finishWait = beginRunPhase("workspace.repository.lock_wait"); + let finishRelease: ((failed?: boolean) => void) | undefined; + let bodyFailed = false; + let bodyError: unknown; + let failed = false; + let releaseFailed = false; + try { + return await withDirectoryPublicationLock(cwd, async () => { + finishWait(); + try { + return await measureRunPhase("workspace.repository.critical_section", work); + } catch (error) { + bodyFailed = true; + bodyError = error; + throw error; + } finally { + // The adapter releases the lock after this callback settles. Keep that + // interval separate from the callback rather than calling it held time. + finishRelease = beginRunPhase("workspace.repository.lock_release"); + } + }); + } catch (error) { + failed = true; + releaseFailed = !bodyFailed || error !== bodyError; + throw error; + } finally { + finishWait(failed); + finishRelease?.(releaseFailed); + } + }); +} + function resolveAcquisitionSource(request: PrepareWorkspaceRepository["repository"], available: ProjectRepository[], anonymousOnly: boolean) { if (anonymousOnly && request.kind === "catalog") throw forbidden(lowTrustRepositoryMessage); return resolveTaskRepository(request, anonymousOnly ? [] : available); @@ -60,28 +94,33 @@ export function assertRepositoryRetryMatches(row: Pick + (await toolAccessService(db).listProjectRepositories(companyId, userId, localTrusted)).repositories); } async function requiresAnonymousAcquisition(reader: Db | DbTransaction, input: { issue: typeof issues.$inferSelect; workspace: Pick | null; agentId?: string | null; runId?: string | null; }) { - const { issue, workspace } = input; - const agentId = input.agentId ?? issue.assigneeAgentId; - const [agent] = agentId ? await reader.select().from(agents).where(and(eq(agents.id, agentId), eq(agents.companyId, issue.companyId))) : []; - if (agentId && !agent) throw forbidden("Task repository agent is unavailable"); - const [project] = issue.projectId ? await reader.select().from(projects).where(and(eq(projects.id, issue.projectId), eq(projects.companyId, issue.companyId))) : []; - const [workspaceSourceProject] = workspace?.projectId ? await reader.select().from(projects).where(and(eq(projects.id, workspace.projectId), eq(projects.companyId, issue.companyId))) : []; - const runId = input.runId ?? issue.executionRunId; - const [run] = runId ? await reader.select(taskRepositoryRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, issue.companyId))) : []; - return requiresAnonymousRepositoryAcquisition({ companyId: issue.companyId, issue, agent, project, - workspaceSourceProject, run, agentId, runId, explicitRunId: input.runId }); + return measureRunPhase("workspace.repository.current_policy", async () => { + const { issue, workspace } = input; + const agentId = input.agentId ?? issue.assigneeAgentId; + const [agent] = agentId ? await measureRunPhase("workspace.repository.db.agent", () => reader.select().from(agents).where(and(eq(agents.id, agentId), eq(agents.companyId, issue.companyId)))) : []; + if (agentId && !agent) throw forbidden("Task repository agent is unavailable"); + const [project] = issue.projectId ? await measureRunPhase("workspace.repository.db.project", () => reader.select().from(projects).where(and(eq(projects.id, issue.projectId!), eq(projects.companyId, issue.companyId)))) : []; + const [workspaceSourceProject] = workspace?.projectId ? await measureRunPhase("workspace.repository.db.project", () => reader.select().from(projects).where(and(eq(projects.id, workspace.projectId!), eq(projects.companyId, issue.companyId)))) : []; + const runId = input.runId ?? issue.executionRunId; + const [run] = runId ? await measureRunPhase("workspace.repository.db.run", () => reader.select(taskRepositoryRunColumns).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, issue.companyId)))) : []; + return requiresAnonymousRepositoryAcquisition({ companyId: issue.companyId, issue, agent, project, + workspaceSourceProject, run, agentId, runId, explicitRunId: input.runId }); + }); } async function assertCurrentControllerAcquisition(input: { companyId: string; issueId: string; workspaceId: string; agentId: string; runId: string }) { - const [issue] = await db.select().from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))); - const [workspace] = await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, input.workspaceId), eq(executionWorkspaces.companyId, input.companyId))); - if (!issue || !workspace || issue.executionWorkspaceId !== workspace.id) throw forbidden("Task repository workspace binding is unavailable"); - if (await requiresAnonymousAcquisition(db, { issue, workspace, agentId: input.agentId, runId: input.runId })) throw forbidden(controllerAcquisitionMessage); + return measureRunPhase("workspace.repository.acquisition", async () => { + const [issue] = await measureRunPhase("workspace.repository.db.issue", () => db.select().from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId)))); + const [workspace] = await measureRunPhase("workspace.repository.db.workspace", () => db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, input.workspaceId), eq(executionWorkspaces.companyId, input.companyId)))); + if (!issue || !workspace || issue.executionWorkspaceId !== workspace.id) throw forbidden("Task repository workspace binding is unavailable"); + if (await requiresAnonymousAcquisition(db, { issue, workspace, agentId: input.agentId, runId: input.runId })) throw forbidden(controllerAcquisitionMessage); + }); } async function request(input: { companyId: string; issueId: string; actor: AuthorizationActor; request: PrepareWorkspaceRepository }) { const parsed = prepareWorkspaceRepositorySchema.parse(input.request); @@ -139,12 +178,12 @@ export function executionWorkspaceRepositoryService(db: Db) { return { kind: "requires_next_admission" as const, applies: "next_normal_admission" as const, operationId: receipt.id, repository: receipt, reason: "immutable_admitted_workspace" }; } async function list(companyId: string, workspaceId: string) { - return db.select().from(executionWorkspaceRepositories).where(and(eq(executionWorkspaceRepositories.companyId, companyId), eq(executionWorkspaceRepositories.executionWorkspaceId, workspaceId))); + return measureRunPhase("workspace.repository.receipts", () => db.select().from(executionWorkspaceRepositories).where(and(eq(executionWorkspaceRepositories.companyId, companyId), eq(executionWorkspaceRepositories.executionWorkspaceId, workspaceId)))); } type PublicationInput = { companyId: string; issueId: string; workspaceId: string; cwd: string; agentId: string; runId: string; responsibleUserId?: string | null }; async function currentRunActor(reader: Db | DbTransaction, input: PublicationInput): Promise { - const [run] = await reader.select(taskRepositoryRunColumns).from(heartbeatRuns).where(and( - eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))); + const [run] = await measureRunPhase("workspace.repository.db.run", () => reader.select(taskRepositoryRunColumns).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId)))); if (!run || run.agentId !== input.agentId) throw forbidden("Task repository run identity is unavailable"); if ("responsibleUserId" in input && run.responsibleUserId !== input.responsibleUserId) { throw forbidden("Task repository responsible user identity is unavailable"); @@ -153,16 +192,19 @@ export function executionWorkspaceRepositoryService(db: Db) { runId: input.runId, onBehalfOfUserId: run.responsibleUserId === "local-board" ? null : run.responsibleUserId }; } async function publicationBinding(reader: Db | DbTransaction, input: PublicationInput, verifyDirectory = true) { - const [issue] = await reader.select().from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))); - const [workspace] = await reader.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, input.workspaceId), eq(executionWorkspaces.companyId, input.companyId))); - if (!issue || !workspace || issue.executionWorkspaceId !== workspace.id || !workspace.cwd - || (verifyDirectory && await canonicalRepositoryRoot(workspace.cwd) !== await canonicalRepositoryRoot(input.cwd))) { - throw forbidden("Task repository publication binding is unavailable"); - } - const acquisitionRestricted = await requiresAnonymousAcquisition(reader, { issue, workspace, agentId: input.agentId, runId: input.runId }); - const actor = await currentRunActor(reader, input); - await assertTaskWorkspaceExecutionAccess(reader, actor, input.companyId, { issueId: issue.id, workspaceId: workspace.id }); - return { issue, workspace, actor, acquisitionRestricted }; + return measureRunPhase("workspace.repository.binding", async () => { + const [issue] = await measureRunPhase("workspace.repository.db.issue", () => reader.select().from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId)))); + const [workspace] = await measureRunPhase("workspace.repository.db.workspace", () => reader.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, input.workspaceId), eq(executionWorkspaces.companyId, input.companyId)))); + if (!issue || !workspace || issue.executionWorkspaceId !== workspace.id || !workspace.cwd + || (verifyDirectory && await measureRunPhase("workspace.repository.paths", async () => + await canonicalRepositoryRoot(workspace.cwd!) !== await canonicalRepositoryRoot(input.cwd)))) { + throw forbidden("Task repository publication binding is unavailable"); + } + const acquisitionRestricted = await requiresAnonymousAcquisition(reader, { issue, workspace, agentId: input.agentId, runId: input.runId }); + const actor = await currentRunActor(reader, input); + await assertTaskWorkspaceExecutionAccess(reader, actor, input.companyId, { issueId: issue.id, workspaceId: workspace.id }); + return { issue, workspace, actor, acquisitionRestricted }; + }); } async function assertPublicationAllowed(input: PublicationInput) { await publicationBinding(db, input); @@ -257,14 +299,14 @@ export function executionWorkspaceRepositoryService(db: Db) { } // The checkout marker names at most one other receipt. This is integrity evidence, // not authority to disclose its originating workspace or task. - const [owner] = await db.select({ repository: executionWorkspaceRepositories, workspace: executionWorkspaces }) + const [owner] = await measureRunPhase("workspace.repository.receipts", () => db.select({ repository: executionWorkspaceRepositories, workspace: executionWorkspaces }) .from(executionWorkspaceRepositories).innerJoin(executionWorkspaces, eq(executionWorkspaces.id, executionWorkspaceRepositories.executionWorkspaceId)) .where(and(eq(executionWorkspaceRepositories.id, ownerId), eq(executionWorkspaceRepositories.companyId, input.row.companyId), - eq(executionWorkspaces.companyId, input.row.companyId))); + eq(executionWorkspaces.companyId, input.row.companyId)))); if (!owner || !owner.workspace.cwd) return pin; const { repository, workspace } = owner; if (repository.repositoryIdentity !== input.row.repositoryIdentity || repository.relativePath !== input.row.relativePath - || await fs.realpath(workspace.cwd!).catch(() => null) !== input.root) return pin; + || await measureRunPhase("workspace.repository.paths", () => fs.realpath(workspace.cwd!).catch(() => null)) !== input.root) return pin; assertRepositoryRetryMatches(repository, input.row.repositoryIdentity, input.row.requestedRef); const markerPin = validPin(input.ownership?.pinnedCommit) ? input.ownership.pinnedCommit : null; if (repository.pinnedCommit && markerPin && repository.pinnedCommit !== markerPin) { @@ -283,63 +325,69 @@ export function executionWorkspaceRepositoryService(db: Db) { if (acquisitionRestricted && rows.some(row => row.state !== "ready" || !row.pinnedCommit || !/^[a-f0-9]{40,64}$/.test(row.pinnedCommit))) { throw forbidden(controllerAcquisitionMessage); } - const root = await fs.realpath(input.cwd); + const root = await measureRunPhase("workspace.repository.paths", () => fs.realpath(input.cwd)); const available = acquisitionRestricted ? [] : await catalog(input.companyId, input.responsibleUserId, input.responsibleUserId === "local-board"); const resolveAuth = createGitRemoteAuthProvider(db, input.companyId, { issueId: input.issueId, heartbeatRunId: input.runId, agentId: input.agentId, responsibleUserId: input.responsibleUserId }); const sources = rows.map(row => { assertAcquisitionReceipt(row, acquisitionRestricted); return resolveAcquisitionSource(row.catalogRepositoryId ? { kind: "catalog", id: row.catalogRepositoryId } : { kind: "url", url: row.repoUrl }, available, acquisitionRestricted); }); - const repositoryRoot = path.join(root, ".paperclip-repositories"); - if (!acquisitionRestricted) await fs.mkdir(repositoryRoot, { recursive: true }); - if ((await fs.lstat(repositoryRoot)).isSymbolicLink() || await fs.realpath(repositoryRoot) !== repositoryRoot) throw conflict("Repository directory escapes the task workspace"); - let stagingParent = root; - if (!acquisitionRestricted) { - for (const name of [".paperclip-runtime", "repository-staging"]) { - stagingParent = path.join(stagingParent, name); - await fs.mkdir(stagingParent, { mode: 0o700 }).catch((error: NodeJS.ErrnoException) => { - if (error.code !== "EEXIST") throw error; - }); - const entry = await fs.lstat(stagingParent); - if (!entry.isDirectory() || entry.isSymbolicLink() || await fs.realpath(stagingParent) !== stagingParent) { - throw conflict("Repository staging directory escapes the task workspace"); + const stagingParent = await measureRunPhase("workspace.repository.fs_prepare", async () => { + const repositoryRoot = path.join(root, ".paperclip-repositories"); + if (!acquisitionRestricted) await fs.mkdir(repositoryRoot, { recursive: true }); + if ((await fs.lstat(repositoryRoot)).isSymbolicLink() || await fs.realpath(repositoryRoot) !== repositoryRoot) throw conflict("Repository directory escapes the task workspace"); + let stagingParent = root; + if (!acquisitionRestricted) { + for (const name of [".paperclip-runtime", "repository-staging"]) { + stagingParent = path.join(stagingParent, name); + await fs.mkdir(stagingParent, { mode: 0o700 }).catch((error: NodeJS.ErrnoException) => { + if (error.code !== "EEXIST") throw error; + }); + const entry = await fs.lstat(stagingParent); + if (!entry.isDirectory() || entry.isSymbolicLink() || await fs.realpath(stagingParent) !== stagingParent) { + throw conflict("Repository staging directory escapes the task workspace"); + } } + const ignorePath = path.join(stagingParent, ".gitignore"); + const ignoreEntry = await fs.lstat(ignorePath).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null; + throw error; + }); + if (ignoreEntry && (!ignoreEntry.isFile() || ignoreEntry.isSymbolicLink())) { + throw conflict("Repository staging ignore file is not a regular file"); + } + const ignoreTemporary = path.join(stagingParent, `.ignore-${randomUUID()}.tmp`); + try { + await fs.writeFile(ignoreTemporary, "*\n", { flag: "wx", mode: 0o600 }); + await fs.rename(ignoreTemporary, ignorePath); + } finally { await fs.rm(ignoreTemporary, { force: true }); } + await measureRunPhase("workspace.repository.ignore_rules", () => ensureManagedRepositoriesIgnored(root)); } - const ignorePath = path.join(stagingParent, ".gitignore"); - const ignoreEntry = await fs.lstat(ignorePath).catch((error: NodeJS.ErrnoException) => { - if (error.code === "ENOENT") return null; - throw error; - }); - if (ignoreEntry && (!ignoreEntry.isFile() || ignoreEntry.isSymbolicLink())) { - throw conflict("Repository staging ignore file is not a regular file"); - } - const ignoreTemporary = path.join(stagingParent, `.ignore-${randomUUID()}.tmp`); - try { - await fs.writeFile(ignoreTemporary, "*\n", { flag: "wx", mode: 0o600 }); - await fs.rename(ignoreTemporary, ignorePath); - } finally { await fs.rm(ignoreTemporary, { force: true }); } - await ensureManagedRepositoriesIgnored(root); - } + return stagingParent; + }); const prepared: Array<{ id: string; cwd: string; repoUrl: string; relativePath: string; pinnedCommit: string; branchName: string | null }> = []; for (const [index, row] of rows.entries()) { const source = sources[index]; if (source.repositoryIdentity !== row.repositoryIdentity || row.relativePath !== taskRepositoryRelativePath(row.repositoryIdentity)) throw conflict("Repository preparation receipt does not match its source"); const cwd = path.join(root, row.relativePath); - await withDirectoryPublicationLock(cwd, async () => { - const existing = await fs.lstat(cwd).catch(() => null); - if (existing && (!existing.isDirectory() || existing.isSymbolicLink() || await fs.realpath(cwd) !== cwd)) throw conflict("Repository checkout path is not a contained directory"); - if (existing) { - const gitDir = path.join(cwd, ".git"); - const metadata = await fs.lstat(gitDir).catch(() => null); - if (!metadata?.isDirectory() || metadata.isSymbolicLink() || await fs.realpath(gitDir) !== gitDir) throw conflict("Repository Git metadata escapes the managed checkout"); - } - if (acquisitionRestricted && !existing) throw forbidden(controllerAcquisitionMessage); + await withRepositoryPublication(cwd, async () => { + const existing = await measureRunPhase("workspace.repository.checkout_identity", async () => { + const existing = await fs.lstat(cwd).catch(() => null); + if (existing && (!existing.isDirectory() || existing.isSymbolicLink() || await fs.realpath(cwd) !== cwd)) throw conflict("Repository checkout path is not a contained directory"); + if (existing) { + const gitDir = path.join(cwd, ".git"); + const metadata = await fs.lstat(gitDir).catch(() => null); + if (!metadata?.isDirectory() || metadata.isSymbolicLink() || await fs.realpath(gitDir) !== gitDir) throw conflict("Repository Git metadata escapes the managed checkout"); + } + if (acquisitionRestricted && !existing) throw forbidden(controllerAcquisitionMessage); + return existing; + }); const anonymous = buildGitAuthInvocation({ token: "", source: "managed_connection", secretName: null }); anonymous.env = { ...anonymous.env, GIT_CONFIG_GLOBAL: "/dev/null", GIT_CONFIG_SYSTEM: "/dev/null" }; - if (!acquisitionRestricted) await db.update(executionWorkspaceRepositories).set({ state: "preparing", catalogRepositoryId: source.catalogRepositoryId, failureCode: null, updatedAt: new Date() }).where(eq(executionWorkspaceRepositories.id, row.id)); + if (!acquisitionRestricted) await measureRunPhase("workspace.repository.state_write", () => db.update(executionWorkspaceRepositories).set({ state: "preparing", catalogRepositoryId: source.catalogRepositoryId, failureCode: null, updatedAt: new Date() }).where(eq(executionWorkspaceRepositories.id, row.id))); try { const ownershipPath = path.join(cwd, ".git", "paperclip-workspace-owner.json"); - const readOwnership = async () => { + const readOwnership = () => measureRunPhase("workspace.repository.ownership_read", async () => { const handle = await fs.open(ownershipPath, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW | fsConstants.O_NONBLOCK).catch(() => null); if (!handle) return null; try { @@ -349,15 +397,15 @@ export function executionWorkspaceRepositoryService(db: Db) { const { bytesRead } = await handle.read(bytes, 0, bytes.length, 0); return bytesRead <= 4096 ? JSON.parse(bytes.subarray(0, bytesRead).toString("utf8")) : null; } catch { return null; } finally { await handle.close(); } - }; + }); const initialOwnership = await readOwnership(); - const authorizedPin = await sharedPublicationPin({ row, root, ownership: initialOwnership }); + const authorizedPin = await measureRunPhase("workspace.repository.ownership_verify", () => sharedPublicationPin({ row, root, ownership: initialOwnership })); if (existing && !authorizedPin && initialOwnership?.repositoryId !== row.id) { throw conflict("Repository directory has no matching preparation receipt; repair it before retrying"); } if (!acquisitionRestricted) { await assertCurrentControllerAcquisition(input); - const result = await materializeManagedProjectWorkspace(cwd, { + const result = await measureRunPhase("workspace.repository.materialize", () => materializeManagedProjectWorkspace(cwd, { repoUrl: source.repoUrl, stagingParent, repoRef: authorizedPin ?? (row.requestedRef === "HEAD" ? null : row.requestedRef), @@ -370,20 +418,21 @@ export function executionWorkspaceRepositoryService(db: Db) { const currentSource = resolveAcquisitionSource({ kind: "catalog", id: source.catalogRepositoryId }, currentCatalog, false); if (currentSource.repositoryIdentity !== row.repositoryIdentity) throw conflict("Repository preparation receipt does not match its source"); } - const pinnedCommit = (await execFile("git", ["-C", cloneCwd, "rev-parse", "HEAD"], { timeout: 30_000 })).stdout.trim(); - await fs.writeFile(path.join(cloneCwd, ".git", "paperclip-workspace-owner.json"), JSON.stringify({ version: 1, repositoryId: row.id, pinnedCommit }), { flag: "wx", mode: 0o600 }); + const pinnedCommit = await measureRunPhase("workspace.repository.git_commit", async () => + (await execFile("git", ["-C", cloneCwd, "rev-parse", "HEAD"], { timeout: 30_000 })).stdout.trim()); + await measureRunPhase("workspace.repository.ownership_write", () => fs.writeFile(path.join(cloneCwd, ".git", "paperclip-workspace-owner.json"), JSON.stringify({ version: 1, repositoryId: row.id, pinnedCommit }), { flag: "wx", mode: 0o600 })); }, - }); + })); if (result.warning) throw conflict(result.warning); } const git = async (...args: string[]) => (await execFile("git", ["-C", cwd, ...args], { timeout: 30_000, env: { ...process.env, GIT_NO_LAZY_FETCH: "1" } })).stdout.trim(); - const origin = normalizeProjectRepositoryUrl(await git("remote", "get-url", "origin")); + const origin = normalizeProjectRepositoryUrl(await measureRunPhase("workspace.repository.git_origin", () => git("remote", "get-url", "origin"))); if (origin.url.toLowerCase() !== row.repositoryIdentity) throw conflict("Repository checkout belongs to a different origin"); const ownership = await readOwnership(); if (!authorizedPin && (ownership?.repositoryId !== row.id || !/^[a-f0-9]{40,64}$/.test(ownership.pinnedCommit))) throw conflict("Repository publication has no valid preparation receipt"); const pinnedCommit: string = authorizedPin ?? ownership.pinnedCommit; - if (acquisitionRestricted) await git("cat-file", "-e", `${pinnedCommit}^{commit}`); - const branchName = await git("symbolic-ref", "--quiet", "--short", "HEAD").catch(() => null); + if (acquisitionRestricted) await measureRunPhase("workspace.repository.git_commit", () => git("cat-file", "-e", `${pinnedCommit}^{commit}`)); + const branchName = await measureRunPhase("workspace.repository.git_branch", () => git("symbolic-ref", "--quiet", "--short", "HEAD").catch(() => null)); await assertPublicationAllowed(input); if (!acquisitionRestricted) { await assertCurrentControllerAcquisition(input); @@ -394,23 +443,25 @@ export function executionWorkspaceRepositoryService(db: Db) { } } if (ownership?.repositoryId !== row.id || ownership?.pinnedCommit !== pinnedCommit) { - const gitDir = path.join(cwd, ".git"); - const metadata = await fs.lstat(gitDir); - if (!metadata.isDirectory() || metadata.isSymbolicLink() || await fs.realpath(gitDir) !== gitDir) { - throw conflict("Repository Git metadata escapes the managed checkout"); - } - const temporaryOwnershipPath = path.join(gitDir, `.paperclip-workspace-owner-${randomUUID()}.tmp`); - try { - await fs.writeFile(temporaryOwnershipPath, JSON.stringify({ version: 1, repositoryId: row.id, pinnedCommit }), { flag: "wx", mode: 0o600 }); - // Keep bounded ownership lookup attached to a receipt for this root, - // even if the workspace that originally published it later moves. - await fs.rename(temporaryOwnershipPath, ownershipPath); - } finally { await fs.rm(temporaryOwnershipPath, { force: true }); } + await measureRunPhase("workspace.repository.ownership_write", async () => { + const gitDir = path.join(cwd, ".git"); + const metadata = await fs.lstat(gitDir); + if (!metadata.isDirectory() || metadata.isSymbolicLink() || await fs.realpath(gitDir) !== gitDir) { + throw conflict("Repository Git metadata escapes the managed checkout"); + } + const temporaryOwnershipPath = path.join(gitDir, `.paperclip-workspace-owner-${randomUUID()}.tmp`); + try { + await fs.writeFile(temporaryOwnershipPath, JSON.stringify({ version: 1, repositoryId: row.id, pinnedCommit }), { flag: "wx", mode: 0o600 }); + // Keep bounded ownership lookup attached to a receipt for this root, + // even if the workspace that originally published it later moves. + await fs.rename(temporaryOwnershipPath, ownershipPath); + } finally { await fs.rm(temporaryOwnershipPath, { force: true }); } + }); } - await db.update(executionWorkspaceRepositories).set({ state: "ready", pinnedCommit, branchName, failureCode: null, updatedAt: new Date() }).where(eq(executionWorkspaceRepositories.id, row.id)); + await measureRunPhase("workspace.repository.state_write", () => db.update(executionWorkspaceRepositories).set({ state: "ready", pinnedCommit, branchName, failureCode: null, updatedAt: new Date() }).where(eq(executionWorkspaceRepositories.id, row.id))); prepared.push({ id: row.id, cwd, repoUrl: row.repoUrl, relativePath: row.relativePath, pinnedCommit, branchName }); } catch (error) { - await db.update(executionWorkspaceRepositories).set({ state: "needs_repair", failureCode: "repository_preparation_failed", updatedAt: new Date() }).where(eq(executionWorkspaceRepositories.id, row.id)); + await measureRunPhase("workspace.repository.state_write", () => db.update(executionWorkspaceRepositories).set({ state: "needs_repair", failureCode: "repository_preparation_failed", updatedAt: new Date() }).where(eq(executionWorkspaceRepositories.id, row.id))); throw error; } }); diff --git a/server/src/services/task-repository-acquisition.ts b/server/src/services/task-repository-acquisition.ts index 876e5b2e827..71c0e07e98b 100644 --- a/server/src/services/task-repository-acquisition.ts +++ b/server/src/services/task-repository-acquisition.ts @@ -4,6 +4,7 @@ import { forbidden } from "../errors.js"; import { isDeepStrictEqual } from "node:util"; import { authorizationService, type AuthorizationActor } from "./authorization.js"; import { LOW_TRUST_ISSUE_ANCESTRY_MAX_DEPTH, resolveCoreTrustPreset } from "./trust-preset-resolver.js"; +import { measureRunPhase } from "./execution-preparation-timing.js"; type DbTransaction = Parameters[0]>[0]; @@ -56,9 +57,9 @@ export async function assertSelectedWorkspaceTrust(db: Db | DbTransaction, input issueId?: string | null; sourceWorkspaceId?: string | null; assigneeAgentId?: string | null; taskCreation?: TaskWorkspaceCreationTrustContext; }) { if (!input.sourceProjectId) return; - const [selectedSource] = await db.select({ id: projects.id, companyId: projects.companyId, + const [selectedSource] = await measureRunPhase("workspace.access.trust.project_read", async () => db.select({ id: projects.id, companyId: projects.companyId, executionWorkspacePolicy: projects.executionWorkspacePolicy }).from(projects).where(and( - eq(projects.companyId, input.companyId), eq(projects.id, input.sourceProjectId))); + eq(projects.companyId, input.companyId), eq(projects.id, input.sourceProjectId!)))); if (!selectedSource) throw forbidden("Selected workspace source project is unavailable"); if (resolveCoreTrustPreset({ companyId: input.companyId, workspaceSourceProject: selectedSource }).kind === "standard") return; const incompatible = () => forbidden("Selected workspace source trust is incompatible with the task's execution policy"); diff --git a/server/src/services/task-workspace-access.ts b/server/src/services/task-workspace-access.ts index 2a8f7b1ece2..609d85a450b 100644 --- a/server/src/services/task-workspace-access.ts +++ b/server/src/services/task-workspace-access.ts @@ -58,75 +58,77 @@ export async function assertTaskWorkspaceExecutionAccess( companyId: string, input: { issueId: string; workspaceId?: string | null; projectWorkspaceId?: string | null; persistedSelection?: TaskWorkspaceSelection }, ) { - if (actor.companyId && actor.companyId !== companyId) throw forbidden("Task workspace belongs to another company"); - const [task] = await db.select({ - id: issues.id, companyId: issues.companyId, projectId: issues.projectId, - executionPolicy: issues.executionPolicy, executionWorkspaceId: issues.executionWorkspaceId, - projectWorkspaceId: issues.projectWorkspaceId, workspaceSelection: issues.workspaceSelection, - workspacePendingSelection: issues.workspacePendingSelection, workspaceBindingRevision: issues.workspaceBindingRevision, - }).from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, companyId))); - if (!task) throw forbidden("Task workspace task is unavailable"); - const access = accessService(db as Db); - const [agent] = actor.type === "agent" && actor.agentId - ? await db.select({ id: agents.id, companyId: agents.companyId, permissions: agents.permissions }) - .from(agents).where(and(eq(agents.id, actor.agentId), eq(agents.companyId, companyId))) : []; - if (actor.type === "agent" && !agent) throw forbidden("Task workspace agent is unavailable"); - const [run] = actor.runId ? await db.select(taskRepositoryRunColumns).from(heartbeatRuns) - .where(and(eq(heartbeatRuns.id, actor.runId), eq(heartbeatRuns.companyId, companyId))) : []; - if ((actor.source === "agent_jwt" && !actor.runId) || (actor.runId && (!run - || run.agentId !== actor.agentId || !runRetainsIssueIdentity(run, task.id)))) { - throw forbidden("Task workspace run identity is unavailable"); - } - if (run && (actor.onBehalfOfUserId ?? null) !== (run.responsibleUserId === "local-board" ? null : run.responsibleUserId)) { - throw forbidden("Task workspace responsible user changed"); - } - const decision = await access.decide({ actor, action: "issue:read", resource: { type: "issue", companyId, issueId: task.id } }); - if (!decision.allowed) throw forbidden(decision.explanation); - const selection = task.workspaceSelection?.selection; - const pending = task.workspacePendingSelection; - const pendingSelection = pending?.expectedBindingRevision === task.workspaceBindingRevision ? pending.intent.selection : undefined; - const effectiveSelection = input.persistedSelection !== undefined - ? taskWorkspaceSelectionSchema.parse(input.persistedSelection) : undefined; - if (effectiveSelection && (input.workspaceId != null || input.projectWorkspaceId != null - || (!isDeepStrictEqual(effectiveSelection, selection) && !isDeepStrictEqual(effectiveSelection, pendingSelection)))) { - throw forbidden("Task workspace selection is not a current persisted intent"); - } - const workspaceId = effectiveSelection - ? effectiveSelection.kind === "existing" ? effectiveSelection.workspaceId : null - : input.workspaceId; - const [workspace] = workspaceId ? await db.select({ - id: executionWorkspaces.id, companyId: executionWorkspaces.companyId, - projectId: executionWorkspaces.projectId, projectWorkspaceId: executionWorkspaces.projectWorkspaceId, - }).from(executionWorkspaces).where(and(eq(executionWorkspaces.id, workspaceId), eq(executionWorkspaces.companyId, companyId))) : []; - if (workspaceId && (!workspace || (task.executionWorkspaceId !== workspace.id - && !(selection?.kind === "existing" && selection.workspaceId === workspace.id) - && !(pendingSelection?.kind === "existing" && pendingSelection.workspaceId === workspace.id)))) { - throw forbidden("Task workspace binding is unavailable"); - } - const selectedSourceId = effectiveSelection - ? effectiveSelection.kind === "configured_source" ? effectiveSelection.projectWorkspaceId : workspace?.projectWorkspaceId ?? null - : input.projectWorkspaceId ?? workspace?.projectWorkspaceId ?? task.projectWorkspaceId; - if (input.projectWorkspaceId && input.projectWorkspaceId !== task.projectWorkspaceId - && input.projectWorkspaceId !== workspace?.projectWorkspaceId - && !(selection?.kind === "configured_source" && selection.projectWorkspaceId === input.projectWorkspaceId) - && !(pendingSelection?.kind === "configured_source" && pendingSelection.projectWorkspaceId === input.projectWorkspaceId)) { - throw forbidden("Task workspace source binding is unavailable"); - } - const [source] = selectedSourceId ? await db.select({ id: projectWorkspaces.id, projectId: projectWorkspaces.projectId }) - .from(projectWorkspaces).where(and(eq(projectWorkspaces.id, selectedSourceId), eq(projectWorkspaces.companyId, companyId))) : []; - if (selectedSourceId && !source) throw forbidden("Task workspace source is unavailable"); - const projectIds = [...new Set([task.projectId, workspace?.projectId, source?.projectId].filter((id): id is string => Boolean(id)))]; - const selectedProjects = projectIds.length ? await db.select({ id: projects.id, companyId: projects.companyId, executionWorkspacePolicy: projects.executionWorkspacePolicy }) - .from(projects).where(and(eq(projects.companyId, companyId), inArray(projects.id, projectIds))) : []; - if (selectedProjects.length !== projectIds.length) throw forbidden("Task workspace project is unavailable"); - for (const projectId of projectIds) await assertTaskWorkspaceSourceProjectAccess(db, actor, companyId, projectId); - const sourceProjectId = source?.projectId ?? workspace?.projectId; - const trust = resolveCoreTrustPreset({ companyId, agent, issue: task, - project: selectedProjects.find(project => project.id === task.projectId), - workspaceSourceProject: selectedProjects.find(project => project.id === sourceProjectId), - run: run ? { companyId: run.companyId, executionPolicy: run.contextSnapshot?.executionPolicy } : null }); - if (trust.kind === "denied") throw forbidden(trust.detail); - await assertSelectedWorkspaceTrust(db, { companyId, actor, issueId: task.id, - sourceProjectId: sourceProjectId ?? null, sourceWorkspaceId: effectiveSelection ? workspace?.id ?? null : workspace?.id }); - return { task, workspace, source }; + return measureRunPhase("workspace.access.execution", async () => { + if (actor.companyId && actor.companyId !== companyId) throw forbidden("Task workspace belongs to another company"); + const [task] = await measureRunPhase("workspace.access.task_read", async () => db.select({ + id: issues.id, companyId: issues.companyId, projectId: issues.projectId, + executionPolicy: issues.executionPolicy, executionWorkspaceId: issues.executionWorkspaceId, + projectWorkspaceId: issues.projectWorkspaceId, workspaceSelection: issues.workspaceSelection, + workspacePendingSelection: issues.workspacePendingSelection, workspaceBindingRevision: issues.workspaceBindingRevision, + }).from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, companyId)))); + if (!task) throw forbidden("Task workspace task is unavailable"); + const access = accessService(db as Db); + const [agent] = actor.type === "agent" && actor.agentId + ? await measureRunPhase("workspace.access.agent_read", async () => db.select({ id: agents.id, companyId: agents.companyId, permissions: agents.permissions }) + .from(agents).where(and(eq(agents.id, actor.agentId!), eq(agents.companyId, companyId)))) : []; + if (actor.type === "agent" && !agent) throw forbidden("Task workspace agent is unavailable"); + const [run] = actor.runId ? await measureRunPhase("workspace.access.run_read", async () => db.select(taskRepositoryRunColumns).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, actor.runId!), eq(heartbeatRuns.companyId, companyId)))) : []; + if ((actor.source === "agent_jwt" && !actor.runId) || (actor.runId && (!run + || run.agentId !== actor.agentId || !runRetainsIssueIdentity(run, task.id)))) { + throw forbidden("Task workspace run identity is unavailable"); + } + if (run && (actor.onBehalfOfUserId ?? null) !== (run.responsibleUserId === "local-board" ? null : run.responsibleUserId)) { + throw forbidden("Task workspace responsible user changed"); + } + const decision = await measureRunPhase("workspace.access.task_authorization", () => access.decide({ actor, action: "issue:read", resource: { type: "issue", companyId, issueId: task.id } })); + if (!decision.allowed) throw forbidden(decision.explanation); + const selection = task.workspaceSelection?.selection; + const pending = task.workspacePendingSelection; + const pendingSelection = pending?.expectedBindingRevision === task.workspaceBindingRevision ? pending.intent.selection : undefined; + const effectiveSelection = input.persistedSelection !== undefined + ? taskWorkspaceSelectionSchema.parse(input.persistedSelection) : undefined; + if (effectiveSelection && (input.workspaceId != null || input.projectWorkspaceId != null + || (!isDeepStrictEqual(effectiveSelection, selection) && !isDeepStrictEqual(effectiveSelection, pendingSelection)))) { + throw forbidden("Task workspace selection is not a current persisted intent"); + } + const workspaceId = effectiveSelection + ? effectiveSelection.kind === "existing" ? effectiveSelection.workspaceId : null + : input.workspaceId; + const [workspace] = workspaceId ? await measureRunPhase("workspace.access.workspace_read", async () => db.select({ + id: executionWorkspaces.id, companyId: executionWorkspaces.companyId, + projectId: executionWorkspaces.projectId, projectWorkspaceId: executionWorkspaces.projectWorkspaceId, + }).from(executionWorkspaces).where(and(eq(executionWorkspaces.id, workspaceId), eq(executionWorkspaces.companyId, companyId)))) : []; + if (workspaceId && (!workspace || (task.executionWorkspaceId !== workspace.id + && !(selection?.kind === "existing" && selection.workspaceId === workspace.id) + && !(pendingSelection?.kind === "existing" && pendingSelection.workspaceId === workspace.id)))) { + throw forbidden("Task workspace binding is unavailable"); + } + const selectedSourceId = effectiveSelection + ? effectiveSelection.kind === "configured_source" ? effectiveSelection.projectWorkspaceId : workspace?.projectWorkspaceId ?? null + : input.projectWorkspaceId ?? workspace?.projectWorkspaceId ?? task.projectWorkspaceId; + if (input.projectWorkspaceId && input.projectWorkspaceId !== task.projectWorkspaceId + && input.projectWorkspaceId !== workspace?.projectWorkspaceId + && !(selection?.kind === "configured_source" && selection.projectWorkspaceId === input.projectWorkspaceId) + && !(pendingSelection?.kind === "configured_source" && pendingSelection.projectWorkspaceId === input.projectWorkspaceId)) { + throw forbidden("Task workspace source binding is unavailable"); + } + const [source] = selectedSourceId ? await measureRunPhase("workspace.access.source_read", async () => db.select({ id: projectWorkspaces.id, projectId: projectWorkspaces.projectId }) + .from(projectWorkspaces).where(and(eq(projectWorkspaces.id, selectedSourceId), eq(projectWorkspaces.companyId, companyId)))) : []; + if (selectedSourceId && !source) throw forbidden("Task workspace source is unavailable"); + const projectIds = [...new Set([task.projectId, workspace?.projectId, source?.projectId].filter((id): id is string => Boolean(id)))]; + const selectedProjects = projectIds.length ? await measureRunPhase("workspace.access.projects_read", async () => db.select({ id: projects.id, companyId: projects.companyId, executionWorkspacePolicy: projects.executionWorkspacePolicy }) + .from(projects).where(and(eq(projects.companyId, companyId), inArray(projects.id, projectIds)))) : []; + if (selectedProjects.length !== projectIds.length) throw forbidden("Task workspace project is unavailable"); + for (const projectId of projectIds) await assertTaskWorkspaceSourceProjectAccess(db, actor, companyId, projectId); + const sourceProjectId = source?.projectId ?? workspace?.projectId; + const trust = resolveCoreTrustPreset({ companyId, agent, issue: task, + project: selectedProjects.find(project => project.id === task.projectId), + workspaceSourceProject: selectedProjects.find(project => project.id === sourceProjectId), + run: run ? { companyId: run.companyId, executionPolicy: run.contextSnapshot?.executionPolicy } : null }); + if (trust.kind === "denied") throw forbidden(trust.detail); + await measureRunPhase("workspace.access.trust", () => assertSelectedWorkspaceTrust(db, { companyId, actor, issueId: task.id, + sourceProjectId: sourceProjectId ?? null, sourceWorkspaceId: effectiveSelection ? workspace?.id ?? null : workspace?.id })); + return { task, workspace, source }; + }); } diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 7fb34b6b7f0..11de5463f08 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -31,6 +31,8 @@ import { } from "./connector-telemetry.js"; import { canBrowseProjectRepositoryGrant, mergeProjectRepository } from "./project-repositories.js"; import { captureRunIdentity } from "./run-identity.js"; +import { measureRunPhase } from "./execution-preparation-timing.js"; +import { measureRunTransaction } from "./execution-transaction-timing.js"; import { createHash, randomBytes, randomUUID } from "node:crypto"; import { readFileSync } from "node:fs"; import { @@ -12141,31 +12143,33 @@ export function toolAccessService( initialGrant: typeof connectionGrants.$inferSelect, actor?: ActorInfo, ) { - try { - return await refreshManagedGitHubGrantAccessOnce( - connection, - initialGrant, - actor, - ); - } catch (error) { - if ( - !(error instanceof HttpError) || - asRecord(error.details).code !== "github_access_changed" - ) - throw error; - const [current] = await db - .select() - .from(connectionGrants) - .where( - and( - eq(connectionGrants.id, initialGrant.id), - eq(connectionGrants.companyId, connection.companyId), - eq(connectionGrants.connectionId, connection.id), - ), + return measureRunPhase("connectors.github.refresh", async () => { + try { + return await refreshManagedGitHubGrantAccessOnce( + connection, + initialGrant, + actor, ); - if (!current || current.status !== "active") throw error; - return refreshManagedGitHubGrantAccessOnce(connection, current, actor); - } + } catch (error) { + if ( + !(error instanceof HttpError) || + asRecord(error.details).code !== "github_access_changed" + ) + throw error; + const [current] = await measureRunPhase("connectors.github.grants", () => db + .select() + .from(connectionGrants) + .where( + and( + eq(connectionGrants.id, initialGrant.id), + eq(connectionGrants.companyId, connection.companyId), + eq(connectionGrants.connectionId, connection.id), + ), + )); + if (!current || current.status !== "active") throw error; + return refreshManagedGitHubGrantAccessOnce(connection, current, actor); + } + }); } async function refreshManagedGitHubGrantAccessOnce( @@ -12173,13 +12177,13 @@ export function toolAccessService( initialGrant: typeof connectionGrants.$inferSelect, actor?: ActorInfo, ) { - let grant = await refreshOAuthGrantCredentials({ + let grant = await measureRunPhase("connectors.github.credentials", () => refreshOAuthGrantCredentials({ companyId: connection.companyId, connectionId: connection.id, grantId: initialGrant.id, actor, - }); - const resolveAccessToken = async () => { + })); + const resolveAccessToken = async () => measureRunPhase("connectors.github.credentials", async () => { const accessRef = grant.credentialSecretRefs.find( (ref) => ref.configPath === "oauth.access_token", ); @@ -12196,15 +12200,15 @@ export function toolAccessService( undefined, ) ).value; - }; + }); let metadata; let accessToken = await resolveAccessToken(); try { - metadata = await loadGitHubGrantMetadata( + metadata = await measureRunPhase("connectors.github.repositories", () => loadGitHubGrantMetadata( accessToken, fetch, grant.providerTenant?.github?.appSlug, - ); + )); } catch (error) { const providerCode = error instanceof HttpError ? asRecord(error.details).code : null; @@ -12212,27 +12216,27 @@ export function toolAccessService( // GitHub may invalidate an access token before its recorded expiry. If an // expiring token pair exists, rotate it once under the same durable lease // and CAS path, then repeat /user before surfacing reconnect-required. - grant = await refreshOAuthGrantCredentials({ + grant = await measureRunPhase("connectors.github.credentials", () => refreshOAuthGrantCredentials({ companyId: connection.companyId, connectionId: connection.id, grantId: initialGrant.id, forceRefresh: true, actor, - }); + })); accessToken = await resolveAccessToken(); try { - metadata = await loadGitHubGrantMetadata( + metadata = await measureRunPhase("connectors.github.repositories", () => loadGitHubGrantMetadata( accessToken, fetch, grant.providerTenant?.github?.appSlug, - ); + )); } catch (retryError) { const retryCode = retryError instanceof HttpError ? asRecord(retryError.details).code : null; if (retryCode === "oauth_reauthorization_required") { - await db + await measureRunPhase("connectors.github.grant_write", () => db .update(connectionGrants) .set({ status: "needs_reauthorization", updatedAt: now() }) .where( @@ -12240,13 +12244,13 @@ export function toolAccessService( eq(connectionGrants.id, grant.id), eq(connectionGrants.companyId, grant.companyId), ), - ); + )); } throw retryError; } } - const { updated, previousGitHub } = await db.transaction(async (tx) => { - const [currentGrant] = await tx + const { updated, previousGitHub } = await measureRunTransaction(db, "connectors.github.refresh", async (tx) => { + const [currentGrant] = await measureRunPhase("connectors.github.grant_lock", () => tx .select() .from(connectionGrants) .where( @@ -12256,7 +12260,7 @@ export function toolAccessService( ), ) .for("update") - .limit(1); + .limit(1)); if (!currentGrant || currentGrant.status === "revoked") throw notFound("GitHub authorization not found"); const previousGitHub = currentGrant.providerTenant?.github; @@ -12284,7 +12288,7 @@ export function toolAccessService( previousGitHub?.webhookHealth ?? metadata.webhookHealth, }, }; - const [updated] = await tx + const [updated] = await measureRunPhase("connectors.github.grant_write", () => tx .update(connectionGrants) .set({ providerTenant, @@ -12297,7 +12301,7 @@ export function toolAccessService( eq(connectionGrants.companyId, grant.companyId), ), ) - .returning(); + .returning()); if (!updated) throw notFound("GitHub authorization not found"); return { updated, previousGitHub }; }); @@ -12312,7 +12316,7 @@ export function toolAccessService( if (cloudConnector && subject) { const previous = new Set(previousGitHub?.installationIds ?? []); const current = new Set(metadata.installationIds); - await Promise.all([ + await measureRunPhase("connectors.github.webhooks", () => Promise.all([ ...metadata.installationIds.map((installationId) => cloudConnector.setWebhookBinding({ subject, @@ -12338,7 +12342,7 @@ export function toolAccessService( active: false, }), ), - ]); + ])); } return updated; } @@ -18260,9 +18264,9 @@ export function toolAccessService( companyId: string, userId: string | null, localTrusted = false, - ) => { + ) => measureRunPhase("connectors.github.catalog", async () => { const [connections, grants, members, memberships] = await Promise.all([ - db + measureRunPhase("connectors.github.connections", () => db .select() .from(toolConnections) .where( @@ -18270,17 +18274,17 @@ export function toolAccessService( eq(toolConnections.companyId, companyId), eq(toolConnections.enabled, true), ), - ), - db + )), + measureRunPhase("connectors.github.grants", () => db .select() .from(connectionGrants) - .where(eq(connectionGrants.companyId, companyId)), - db + .where(eq(connectionGrants.companyId, companyId))), + measureRunPhase("connectors.github.grant_members", () => db .select() .from(connectionGrantMembers) - .where(eq(connectionGrantMembers.companyId, companyId)), + .where(eq(connectionGrantMembers.companyId, companyId))), userId - ? db + ? measureRunPhase("connectors.github.membership", () => db .select() .from(companyMemberships) .where( @@ -18290,7 +18294,7 @@ export function toolAccessService( eq(companyMemberships.principalId, userId), eq(companyMemberships.status, "active"), ), - ) + )) : Promise.resolve([]), ]); const repositories = new Map< @@ -18357,8 +18361,8 @@ export function toolAccessService( ); rows = grant.providerTenant?.github?.repositories ?? []; } else { - const headers = initialGrant - ? await (async () => { + const headers = await measureRunPhase("connectors.github.credentials", () => initialGrant + ? (async () => { const ref = initialGrant.credentialSecretRefs.find( (ref) => ref.configPath === "oauth.access_token" || @@ -18377,8 +18381,8 @@ export function toolAccessService( ); return { Authorization: `Bearer ${secret.value}` }; })() - : await resolveCredentialHeaders(connection, actor); - rows = await loadGitHubTokenRepositories(headers); + : resolveCredentialHeaders(connection, actor)); + rows = await measureRunPhase("connectors.github.repositories", () => loadGitHubTokenRepositories(headers)); } for (const row of rows) { mergeProjectRepository(repositories, row, connection.name, connection.id); @@ -18399,7 +18403,7 @@ export function toolAccessService( connectionCount, failedConnectionCount, }; - }, + }), listConnections: async ( companyId: string,