Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-08 01:57:31 +01:00
Code Issues Packages Projects Releases Wiki Activity

trace(server): expose repository admission waits (#17040)

Browse Source 
## 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-20 19:50:51 -05:00
1 parent 8b4ab65b1c
commit 6c9b87d092
10 files changed
+989 -297
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/observability.md
+118 -7
View File
@@ -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
server/src/__tests__/github-catalog-timing.test.ts
+198
View File
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
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<void>(resolve => { release = resolve; });
const ready = new Promise<void>(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);
});
});
server/src/__tests__/repository-admission-tracing.test.ts
+285
View File
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
let directory: string;
let fixture: Awaited<ReturnType<typeof createWorkspacePreparationHistoryFixture>>;
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<void>) {
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<void>(resolve => { release = resolve; });
const ready = new Promise<void>(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);
});
});
server/src/services/authorization.ts
+76 -62
View File
@@ -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<T>(phase: Extract<ExecutionPhase, `authorization.${string}`>, work: () => T | Promise<T>): T | Promise<T> {
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<string, unknown> | null;
}): Promise<AuthorizationDecision> {
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
server/src/services/execution-preparation-timing.ts
+28 -2
View File
@@ -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<string, number> };
type RunScope = { runId: string; phase: string; parent?: Context; ended: boolean; details: Map<string, number> };
const ownership = new AsyncLocalStorage<RunScope>();
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<T>(runId: string, name: string, work: () => T | Promise<T>, independent = false, counts: { fileCount?: number; bytes?: number } = {}): Promise<T> {
let operation: Promise<T> | 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<string, number>(),
server/src/services/execution-transaction-timing.ts
+1 -1
View File
@@ -3,7 +3,7 @@ import { beginRunPhase, measureRunPhase } from "./execution-preparation-timing.j
type Transaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
export function measureRunTransaction<T>(db: Db, scope: "workspace" | "runtime_selection" | "response" | "connectors.slack.authority" | "connectors.slack.catalog", work: (tx: Transaction) => Promise<T>): Promise<T> {
export function measureRunTransaction<T>(db: Db, scope: "workspace" | "runtime_selection" | "response" | "connectors.slack.authority" | "connectors.slack.catalog" | "connectors.github.refresh", work: (tx: Transaction) => Promise<T>): Promise<T> {
return measureRunPhase(`${scope}.transaction`, async () => {
const entry = beginRunPhase(`${scope}.transaction.entry`);
let settle: ((failed?: boolean) => void) | undefined;
server/src/services/execution-workspace-repositories.ts
+146 -95
View File
@@ -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<Parameters<Db["transaction"]>[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<T>(cwd: string, work: () => Promise<T>): Promise<T> {
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<RepositoryRow, "repositor
export function executionWorkspaceRepositoryService(db: Db) {
async function catalog(companyId: string, userId: string | null, localTrusted: boolean) {
return (await toolAccessService(db).listProjectRepositories(companyId, userId, localTrusted)).repositories;
return measureRunPhase("workspace.repository.catalog", async () =>
(await toolAccessService(db).listProjectRepositories(companyId, userId, localTrusted)).repositories);
}
async function requiresAnonymousAcquisition(reader: Db | DbTransaction, input: {
issue: typeof issues.$inferSelect; workspace: Pick<typeof executionWorkspaces.$inferSelect, "projectId"> | 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<AuthorizationActor> {
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;
}
});
server/src/services/task-repository-acquisition.ts
+3 -2
View File
@@ -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<Parameters<Db["transaction"]>[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");
server/src/services/task-workspace-access.ts
+73 -71
View File
@@ -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 };
});
}
server/src/services/tool-access.ts
+61 -57
View File
@@ -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,