fix(hub): check a document changed during approval on the hub's clock

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",

apps/hub/src/artifacts/service.ts

@@ -2141,11 +2141,15 @@ 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,
-            askedAt: approval.createdAt,
+            askedAgoMs: sql<number>`extract(epoch from now()::timestamp - ${approval.createdAt})::float8 * 1000`,
             askedKinds: sql<string | null>`${approval.toolArguments}->>'kinds'`,
           })
           .from(approval)
@@ -2183,8 +2187,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 = checkedAt - 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", {

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`), 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`).

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<ReturnType<typeof createTestDatabase>>;
+let handle: ReturnType<typeof createDB>;
+let sql: postgres.Sql;
+let service: ReturnType<typeof createArtifactService>;
+
+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<typeof createArtifactService>[1],
+  );
+});
+
+afterAll(async () => {
+  await sql?.end();
+  await handle?.close();
+  await database?.drop();
+});
+
+type Scope = { tenantId: string; principalId: string; projectId: string };
+
+async function seed(): Promise<Scope> {
+  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<T>(skewMs: number, step: () => Promise<T>) {
+  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<ReturnType<typeof file>>,
+  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: [],
+    });
+  });
+}

test/e2e/support/hub.ts

@@ -74,27 +74,47 @@ async function signUpSessionCookie(
   return cookies.join("; ");
 }
 
-export async function startTestHub(
-  recordedModel: RecordedModelServer,
-): Promise<TestHubHandle> {
+// 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<string, string>;
+  drop: () => Promise<void>;
+}> {
   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}"`);
-  } finally {
-    await adminSql.end();
-  }
+export async function startTestHub(
+  recordedModel: RecordedModelServer,
+): Promise<TestHubHandle> {
+  const database = await createTestDatabase();
 
   const dataDir = await mkdtemp(join(tmpdir(), "builder-e2e-hub-"));
   const fakeCredentialSecret = `sk-fake-e2e-secret-${crypto.randomUUID()}`;
@@ -117,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,
@@ -144,23 +159,6 @@ export async function startTestHub(
   let server: ReturnType<typeof Bun.serve> | undefined;
   const activeStreams = new Set<AbortController>();
 
-  async function dropDatabase(): Promise<void> {
-    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);
@@ -319,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) {
@@ -330,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;
   }
 }