From 133cb40b0e7a236016d1c4d08ff8744ec0240080 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 15:50:55 -0800 Subject: [PATCH 1/2] fix(hub): check a document changed during approval on the hub's clock --- apps/hub/src/artifacts/service.ts | 7 +++++-- test/AGENTS.md | 3 ++- test/e2e/support/hub.ts | 23 +++++++++++++++++++++++ 3 files changed, 30 insertions(+), 3 deletions(-) diff --git a/apps/hub/src/artifacts/service.ts b/apps/hub/src/artifacts/service.ts index 5641ce30..3b8db40a 100644 --- a/apps/hub/src/artifacts/service.ts +++ b/apps/hub/src/artifacts/service.ts @@ -2145,7 +2145,9 @@ export function createArtifactService( const [granted] = await tx .select({ id: approval.id, - askedAt: approval.createdAt, + // The database stamps the ask and the hub stamps versions, so the ask's + // age moves it onto the hub's clock. + askedAgoMs: sql`extract(epoch from now()::timestamp - ${approval.createdAt})::float8 * 1000`, askedKinds: sql`${approval.toolArguments}->>'kinds'`, }) .from(approval) @@ -2183,8 +2185,9 @@ export function createArtifactService( if (recorded || !artifacts.length) return { approvalId: granted.id, artifacts: [] }; // A version written while the person was deciding is not one they saw. + const askedAt = Date.now() - granted.askedAgoMs; const changed = artifacts.find( - ({ writtenAt }) => writtenAt > granted.askedAt, + ({ writtenAt }) => writtenAt.getTime() > askedAt, )?.ref; if (changed) throw new ProjectError(409, "artifact_changed_during_approval", { diff --git a/test/AGENTS.md b/test/AGENTS.md index 2e9fdbcf..73857bc2 100644 --- a/test/AGENTS.md +++ b/test/AGENTS.md @@ -9,7 +9,8 @@ end-to-end tests: `bun run test` with no database. - **`test/e2e/`:** end-to-end tests against an isolated Postgres, the real hub and real sidecars (`custom-provider.test.ts`, `oauth-model-login.test.ts`, - `specialist-end-notices.test.ts` and `pipeline.test.ts`). + `specialist-end-notices.test.ts` and `pipeline.test.ts`). Each test + database's `now()` runs five minutes behind the hub's clock. - **`test/e2e/support/`:** shared harness (`hub.ts`), the recorded OpenAI-compatible model server (`recorded-model.ts` and its unit tests in `recorded-model.test.ts`) and the fixture extractor (`extract.ts`). diff --git a/test/e2e/support/hub.ts b/test/e2e/support/hub.ts index 24b5aab7..1efc327e 100644 --- a/test/e2e/support/hub.ts +++ b/test/e2e/support/hub.ts @@ -23,6 +23,8 @@ export type TestHubHandle = Readonly<{ close: () => Promise; }>; +const DB_CLOCK_LAG = "5 minutes"; + function readTestDbServerConfig(): { host: string; port: number; @@ -92,9 +94,30 @@ export async function startTestHub( try { await adminSql.unsafe(`CREATE DATABASE "${dbName}"`); + // The database clock runs behind the hub's, as a separately hosted database's can, so + // nothing passes only because both stamp time from one machine's clock. + await adminSql.unsafe( + `ALTER DATABASE "${dbName}" SET search_path = "$user", public, pg_catalog`, + ); } finally { await adminSql.end(); } + const lagSql = postgres({ + host: dbServer.host, + port: dbServer.port, + user: dbServer.user, + password: dbServer.password, + database: dbName, + max: 1, + onnotice: () => undefined, + }); + try { + await lagSql.unsafe( + `CREATE FUNCTION public.now() RETURNS timestamptz LANGUAGE sql STABLE AS $$ SELECT pg_catalog.now() - interval '${DB_CLOCK_LAG}' $$`, + ); + } finally { + await lagSql.end(); + } const dataDir = await mkdtemp(join(tmpdir(), "builder-e2e-hub-")); const fakeCredentialSecret = `sk-fake-e2e-secret-${crypto.randomUUID()}`; From 5c28ab6aa3ba0104eb92000646c18b68276e4af1 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 16:26:06 -0800 Subject: [PATCH 2/2] fix(hub): err toward refusing; test approval check under clock skew --- apps/hub/package.json | 1 + apps/hub/src/artifacts/service.ts | 8 +- test/AGENTS.md | 4 +- test/e2e/approval-clock.test.ts | 172 ++++++++++++++++++++++++++++++ test/e2e/support/hub.ts | 107 +++++++------------ 5 files changed, 221 insertions(+), 71 deletions(-) create mode 100644 test/e2e/approval-clock.test.ts diff --git a/apps/hub/package.json b/apps/hub/package.json index 07a1654a..17d0a4d8 100644 --- a/apps/hub/package.json +++ b/apps/hub/package.json @@ -7,6 +7,7 @@ "type": "module", "exports": { "./app": "./src/app.ts", + "./artifacts/service": "./src/artifacts/service.ts", "./environment": "./src/environment.ts", "./migrate": "./src/migrate.ts", "./workflows/install-for-tenant": "./src/workflows/install-for-tenant.ts", diff --git a/apps/hub/src/artifacts/service.ts b/apps/hub/src/artifacts/service.ts index 3b8db40a..eb53553c 100644 --- a/apps/hub/src/artifacts/service.ts +++ b/apps/hub/src/artifacts/service.ts @@ -2141,12 +2141,14 @@ export function createArtifactService( }); artifacts.push({ ref, writtenAt: new Date(newest.createdAt) }); } + // The database stamps the ask and the hub stamps versions, so the ask's age at the + // transaction's start moves it onto the hub's clock. Reading the hub's clock first + // places the ask no later than it was, so any doubt refuses. + const checkedAt = Date.now(); return rawDB.transaction(async (tx) => { const [granted] = await tx .select({ id: approval.id, - // The database stamps the ask and the hub stamps versions, so the ask's - // age moves it onto the hub's clock. askedAgoMs: sql`extract(epoch from now()::timestamp - ${approval.createdAt})::float8 * 1000`, askedKinds: sql`${approval.toolArguments}->>'kinds'`, }) @@ -2185,7 +2187,7 @@ export function createArtifactService( if (recorded || !artifacts.length) return { approvalId: granted.id, artifacts: [] }; // A version written while the person was deciding is not one they saw. - const askedAt = Date.now() - granted.askedAgoMs; + const askedAt = checkedAt - granted.askedAgoMs; const changed = artifacts.find( ({ writtenAt }) => writtenAt.getTime() > askedAt, )?.ref; diff --git a/test/AGENTS.md b/test/AGENTS.md index 73857bc2..2f616291 100644 --- a/test/AGENTS.md +++ b/test/AGENTS.md @@ -9,8 +9,8 @@ end-to-end tests: `bun run test` with no database. - **`test/e2e/`:** end-to-end tests against an isolated Postgres, the real hub and real sidecars (`custom-provider.test.ts`, `oauth-model-login.test.ts`, - `specialist-end-notices.test.ts` and `pipeline.test.ts`). Each test - database's `now()` runs five minutes behind the hub's clock. + `specialist-end-notices.test.ts` and `pipeline.test.ts`), and hub services + driven in-process against that Postgres (`approval-clock.test.ts`). - **`test/e2e/support/`:** shared harness (`hub.ts`), the recorded OpenAI-compatible model server (`recorded-model.ts` and its unit tests in `recorded-model.test.ts`) and the fixture extractor (`extract.ts`). diff --git a/test/e2e/approval-clock.test.ts b/test/e2e/approval-clock.test.ts new file mode 100644 index 00000000..d27418f4 --- /dev/null +++ b/test/e2e/approval-clock.test.ts @@ -0,0 +1,172 @@ +import { afterAll, beforeAll, expect, setSystemTime, test } from "bun:test"; +import postgres from "postgres"; +import { createArtifact, writeArtifactVersion } from "@corbits/artifacts"; +import { createDB } from "@intx/db"; +import { createArtifactService } from "@corbits-builder/hub/artifacts/service"; +import { readDatabaseConfig } from "@corbits-builder/hub/environment"; +import { migrateDatabase } from "@corbits-builder/hub/migrate"; +import { createTestDatabase } from "./support/hub.ts"; + +// The database stamps an ask and the hub stamps a document's versions; their clocks disagree +// by this much, either way. +const SKEW_MS = 5 * 60 * 1000; + +let database: Awaited>; +let handle: ReturnType; +let sql: postgres.Sql; +let service: ReturnType; + +beforeAll(async () => { + database = await createTestDatabase(); + await migrateDatabase(database.env); + const config = readDatabaseConfig(database.env); + handle = createDB(config); + sql = postgres({ + host: config.host, + port: config.port, + user: config.user, + password: config.password, + database: config.database, + max: 1, + onnotice: () => undefined, + }); + service = createArtifactService( + handle.db, + {} as Parameters[1], + ); +}); + +afterAll(async () => { + await sql?.end(); + await handle?.close(); + await database?.drop(); +}); + +type Scope = { tenantId: string; principalId: string; projectId: string }; + +async function seed(): Promise { + const id = crypto.randomUUID().replaceAll("-", "").slice(0, 16); + const scope = { + tenantId: `tnt_${id}`, + principalId: `prn_${id}`, + projectId: `project-${id}`, + }; + await sql`insert into tenant (id, name, slug, domain) values (${scope.tenantId}, ${id}, ${id}, ${`${id}.localhost`})`; + await sql`insert into principal (id, tenant_id, kind, ref_id, status) values (${scope.principalId}, ${scope.tenantId}, 'user', ${id}, 'active')`; + await sql`insert into "grant" (id, tenant_id, principal_id, resource, action, effect, origin) values (${`grt_${id}`}, ${scope.tenantId}, ${scope.principalId}, ${`sb-project:${scope.projectId}`}, 'manage', 'allow', 'system')`; + await sql`insert into project (id, tenant_id, title) values (${scope.projectId}, ${scope.tenantId}, ${id})`; + await sql`insert into workflow_definition (id, tenant_id, name) values (${`wfd_${id}`}, ${scope.tenantId}, ${id})`; + await sql`insert into workflow_run (id, tenant_id, definition_id) values (${`run_${id}`}, ${scope.tenantId}, ${`wfd_${id}`})`; + return scope; +} + +// Runs a hub step with the hub's clock skewed from the database's. +async function onHubClock(skewMs: number, step: () => Promise) { + setSystemTime(); + setSystemTime(new Date(Date.now() + skewMs)); + try { + return await step(); + } finally { + setSystemTime(); + } +} + +// Keeps the steps' wall-clock order distinct at millisecond resolution. +const pause = () => Bun.sleep(20); + +async function file(scope: Scope, skewMs: number) { + const row = await onHubClock(skewMs, () => + handle.db.transaction((tx) => + createArtifact(tx, { + scope, + ownerPrincipalId: scope.principalId, + kind: "problem_brief", + title: "Brief", + content: "first", + source: {}, + }), + ), + ); + return { + artifactId: row.id, + kind: "problem_brief", + version: row.version, + sha256: row.contentSha256!, + }; +} + +async function revise(scope: Scope, artifactId: string, skewMs: number) { + const row = await onHubClock(skewMs, () => + writeArtifactVersion(handle.db, { scope, artifactId, content: "second" }), + ); + return { + artifactId, + kind: "problem_brief", + version: row.version, + sha256: row.contentSha256!, + }; +} + +// The person is asked and approves; the database stamps when they were asked. +async function askAndApprove(scope: Scope) { + const approvalId = `apr_${crypto.randomUUID()}`; + const runId = `run_${scope.tenantId.slice(4)}`; + await sql`insert into approval (id, tenant_id, anchor_run_id, run_id, agent_address, correlation_id, tool_definition, tool_arguments, status, resolved_at) + values (${approvalId}, ${scope.tenantId}, ${runId}, ${runId}, 'coordinator@localhost', ${approvalId}, ${sql.json({ name: "request_approval" })}, ${sql.json({ kinds: ["problem_brief"] })}, 'approved', now())`; + return approvalId; +} + +function record( + scope: Scope, + ref: Awaited>, + skewMs: number, +) { + return onHubClock(skewMs, () => + service.recordApproval(scope, scope.projectId, { + tool: "request_approval", + artifacts: [ref], + }), + ); +} + +for (const skewMs of [SKEW_MS, -SKEW_MS]) { + const clocks = `with the hub's clock ${skewMs > 0 ? "ahead of" : "behind"} the database's`; + + test(`${clocks}, a document written before the ask is approved`, async () => { + const scope = await seed(); + const ref = await file(scope, skewMs); + await pause(); + const approvalId = await askAndApprove(scope); + await pause(); + expect(await record(scope, ref, skewMs)).toEqual({ + approvalId, + artifacts: [ref], + }); + }); + + test(`${clocks}, a document revised after the ask is refused`, async () => { + const scope = await seed(); + const first = await file(scope, skewMs); + await pause(); + await askAndApprove(scope); + await pause(); + const revised = await revise(scope, first.artifactId, skewMs); + await pause(); + await expect(record(scope, revised, skewMs)).rejects.toMatchObject({ + code: "artifact_changed_during_approval", + }); + }); + + test(`${clocks}, an approval already recorded records nothing more`, async () => { + const scope = await seed(); + const ref = await file(scope, skewMs); + await pause(); + const approvalId = await askAndApprove(scope); + await pause(); + await record(scope, ref, skewMs); + expect(await record(scope, ref, skewMs)).toEqual({ + approvalId, + artifacts: [], + }); + }); +} diff --git a/test/e2e/support/hub.ts b/test/e2e/support/hub.ts index 1efc327e..2d385a15 100644 --- a/test/e2e/support/hub.ts +++ b/test/e2e/support/hub.ts @@ -23,8 +23,6 @@ export type TestHubHandle = Readonly<{ close: () => Promise; }>; -const DB_CLOCK_LAG = "5 minutes"; - function readTestDbServerConfig(): { host: string; port: number; @@ -76,48 +74,47 @@ async function signUpSessionCookie( return cookies.join("; "); } -export async function startTestHub( - recordedModel: RecordedModelServer, -): Promise { +// An empty database of its own on the e2e server, with the hub's DB_* settings for it. +export async function createTestDatabase(): Promise<{ + env: Record; + drop: () => Promise; +}> { const dbServer = readTestDbServerConfig(); const dbName = `builder_e2e_${crypto.randomUUID().replaceAll("-", "").slice(0, 16)}`; + const admin = async (statement: string) => { + const adminSql = postgres({ + host: dbServer.host, + port: dbServer.port, + user: dbServer.user, + password: dbServer.password, + database: dbServer.adminDatabase, + max: 1, + onnotice: () => undefined, + }); + try { + await adminSql.unsafe(statement); + } finally { + await adminSql.end(); + } + }; + await admin(`CREATE DATABASE "${dbName}"`); + return { + env: { + DB_HOST: dbServer.host, + DB_PORT: String(dbServer.port), + DB_USER: dbServer.user, + DB_PASSWORD: dbServer.password, + DB_NAME: dbName, + DB_SSL: "false", + }, + drop: () => admin(`DROP DATABASE IF EXISTS "${dbName}" WITH (FORCE)`), + }; +} - const adminSql = postgres({ - host: dbServer.host, - port: dbServer.port, - user: dbServer.user, - password: dbServer.password, - database: dbServer.adminDatabase, - max: 1, - onnotice: () => undefined, - }); - - try { - await adminSql.unsafe(`CREATE DATABASE "${dbName}"`); - // The database clock runs behind the hub's, as a separately hosted database's can, so - // nothing passes only because both stamp time from one machine's clock. - await adminSql.unsafe( - `ALTER DATABASE "${dbName}" SET search_path = "$user", public, pg_catalog`, - ); - } finally { - await adminSql.end(); - } - const lagSql = postgres({ - host: dbServer.host, - port: dbServer.port, - user: dbServer.user, - password: dbServer.password, - database: dbName, - max: 1, - onnotice: () => undefined, - }); - try { - await lagSql.unsafe( - `CREATE FUNCTION public.now() RETURNS timestamptz LANGUAGE sql STABLE AS $$ SELECT pg_catalog.now() - interval '${DB_CLOCK_LAG}' $$`, - ); - } finally { - await lagSql.end(); - } +export async function startTestHub( + recordedModel: RecordedModelServer, +): Promise { + const database = await createTestDatabase(); const dataDir = await mkdtemp(join(tmpdir(), "builder-e2e-hub-")); const fakeCredentialSecret = `sk-fake-e2e-secret-${crypto.randomUUID()}`; @@ -140,12 +137,7 @@ export async function startTestHub( HOME: process.env["HOME"] ?? tmpdir(), USER: process.env["USER"] ?? "e2e", TMPDIR: process.env["TMPDIR"] ?? tmpdir(), - DB_HOST: dbServer.host, - DB_PORT: String(dbServer.port), - DB_USER: dbServer.user, - DB_PASSWORD: dbServer.password, - DB_NAME: dbName, - DB_SSL: "false", + ...database.env, HOST: "127.0.0.1", PORT: String(port), BETTER_AUTH_BASE_URL: baseUrl, @@ -167,23 +159,6 @@ export async function startTestHub( let server: ReturnType | undefined; const activeStreams = new Set(); - async function dropDatabase(): Promise { - const dropSql = postgres({ - host: dbServer.host, - port: dbServer.port, - user: dbServer.user, - password: dbServer.password, - database: dbServer.adminDatabase, - max: 1, - onnotice: () => undefined, - }); - try { - await dropSql.unsafe(`DROP DATABASE IF EXISTS "${dbName}" WITH (FORCE)`); - } finally { - await dropSql.end(); - } - } - try { await migrateDatabase(hubEnv); runtime = await createHubRuntime(hubEnv); @@ -342,7 +317,7 @@ export async function startTestHub( if (server) await server.stop(true); if (runtime) await runtime.close(); await rm(dataDir, { recursive: true, force: true }); - await dropDatabase(); + await database.drop(); }, }; } catch (error) { @@ -353,7 +328,7 @@ export async function startTestHub( if (server) await server.stop(true); if (runtime) await runtime.close(); await rm(dataDir, { recursive: true, force: true }); - await dropDatabase(); + await database.drop(); throw error; } }