feat(agentd): support configured db schema

docker-compose.yml

@@ -126,6 +126,7 @@ services:
       AGENTD_AUTHORITY_STORE: ${AGENTD_AUTHORITY_STORE:-postgres}
       AGENTD_DATABASE_URL: ${AGENTD_DATABASE_URL:-postgresql://${POSTGRES_USER:-fides}:${POSTGRES_PASSWORD:?Set POSTGRES_PASSWORD in .env}@postgres:5432/${POSTGRES_DB:-fides}}
       AGENTD_DB_AUTO_MIGRATE: ${AGENTD_DB_AUTO_MIGRATE:-true}
+      AGENTD_DB_SCHEMA: ${AGENTD_DB_SCHEMA:-agentd}
       AGENTD_DB_POOL_MAX: ${AGENTD_DB_POOL_MAX:-10}
       DISCOVERY_URL: http://discovery:3100
       TRUST_GRAPH_URL: http://trust-graph:3200

docs/deployment.md

@@ -60,6 +60,7 @@ cp .env.example .env
 | `AGENTD_AUTHORITY_STORE`  | `file`  | production | `file` for local JSON state, `postgres` for durable authority state |
 | `AGENTD_DATABASE_URL`     | _(empty)_ | production when `AGENTD_AUTHORITY_STORE=postgres` | Dedicated agentd authority database URL. Falls back to `DATABASE_URL` when unset. |
 | `AGENTD_DB_AUTO_MIGRATE`  | `true`  | no | Runs idempotent authority migrations on startup and records applied ids in `agentd_schema_migrations`. Set `false` when migrations are managed externally. |
+| `AGENTD_DB_SCHEMA`        | `agentd` in Docker Compose, otherwise _(empty)_ | no | Optional Postgres schema name for agentd authority tables when multiple services share one database. Must be a simple identifier. |
 | `AGENTD_DB_POOL_MAX`      | `10`    | no | Agentd authority store connection pool size. Falls back to `DB_POOL_MAX`. |
 | `AGENTD_STATE_STORE_PATH` | _(empty)_ | no | File authority store path. Defaults to `~/.fides/agentd/authority-store.json`. |
 | `AGENTD_REQUIRE_AUTHORITY_SIGNATURE_VERIFICATION` | `true` in production, `false` otherwise | no | When `true`, agentd rejects delegation, revocation, and incident writes unless the request includes the corresponding signer public key for canonical signature verification. Set `false` only for transitional deployments that cannot yet send signer public keys. |

services/agentd/src/migrate.ts

@@ -1,17 +1,12 @@
-import postgres from 'postgres'
-import { runAuthorityMigrations } from './storage.js'
+import { createAuthorityClient, runAuthorityMigrations } from './storage.js'
 
 async function main() {
   const connectionString = process.env.AGENTD_DATABASE_URL || process.env.DATABASE_URL
   if (!connectionString) {
     throw new Error('AGENTD_DATABASE_URL or DATABASE_URL is required')
   }
 
-  const sql = postgres(connectionString, {
-    max: 1,
-    idle_timeout: 5,
-    connect_timeout: 10,
-  })
+  const sql = createAuthorityClient(connectionString)
 
   try {
     await runAuthorityMigrations(sql)

services/agentd/src/storage.ts

@@ -322,11 +322,7 @@ export class PostgresAuthorityStore implements AuthorityStore {
   private initialized: Promise<void>
 
   constructor(connectionString = requiredDatabaseUrl()) {
-    this.sql = postgres(connectionString, {
-      max: parseInt(process.env.AGENTD_DB_POOL_MAX || process.env.DB_POOL_MAX || '10', 10),
-      idle_timeout: 20,
-      connect_timeout: 10,
-    })
+    this.sql = createAuthorityClient(connectionString)
     this.initialized = this.init()
   }
 
@@ -497,6 +493,7 @@ export async function runAuthorityMigrations(sql: postgres.Sql): Promise<void> {
   await sql`SELECT pg_advisory_lock(hashtext('agentd_authority_migrations'))`
 
   try {
+    await ensureConfiguredAuthoritySchema(sql)
     await ensureAuthorityMigrationLedger(sql)
     for (const migration of AUTHORITY_MIGRATIONS) {
       const checksum = authorityMigrationChecksum(migration)
@@ -532,6 +529,14 @@ export async function runAuthorityMigrations(sql: postgres.Sql): Promise<void> {
   }
 }
 
+async function ensureConfiguredAuthoritySchema(sql: postgres.Sql): Promise<void> {
+  const schemaName = process.env.AGENTD_DB_SCHEMA
+  if (!schemaName) return
+
+  assertValidAuthoritySchemaName(schemaName)
+  await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS "${schemaName}"`)
+}
+
 async function ensureAuthorityMigrationLedger(sql: postgres.Sql): Promise<void> {
   await sql`
     CREATE TABLE IF NOT EXISTS agentd_schema_migrations (
@@ -629,3 +634,27 @@ function requiredDatabaseUrl(): string {
   }
   return url
 }
+
+export function createAuthorityClient(connectionString = requiredDatabaseUrl()): postgres.Sql {
+  return postgres(withAuthoritySearchPath(connectionString, process.env.AGENTD_DB_SCHEMA), {
+    max: parseInt(process.env.AGENTD_DB_POOL_MAX || process.env.DB_POOL_MAX || '10', 10),
+    idle_timeout: 20,
+    connect_timeout: 10,
+  })
+}
+
+function withAuthoritySearchPath(connectionString: string, schemaName?: string): string {
+  if (!schemaName) return connectionString
+
+  assertValidAuthoritySchemaName(schemaName)
+
+  const url = new URL(connectionString)
+  url.searchParams.set('options', `-c search_path=${schemaName},public`)
+  return url.toString()
+}
+
+function assertValidAuthoritySchemaName(schemaName: string): void {
+  if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(schemaName)) {
+    throw new Error('AGENTD_DB_SCHEMA must be a simple Postgres identifier')
+  }
+}

services/agentd/test/storage.test.ts

@@ -10,6 +10,7 @@ import {
   FileAuthorityStore,
   InMemoryAuthorityStore,
   PostgresAuthorityStore,
+  createAuthorityClient,
   runAuthorityMigrations,
 } from '../src/storage.js'
 
@@ -138,6 +139,23 @@ describe('agentd authority stores', () => {
     expect((await store.getSession(grant.id))?.revocationReason).toBe('manual')
   })
 
+  it('rejects unsafe configured authority schema names', () => {
+    const previousSchema = process.env.AGENTD_DB_SCHEMA
+
+    try {
+      process.env.AGENTD_DB_SCHEMA = 'agentd;DROP'
+      expect(() => createAuthorityClient('postgresql://fides:fides@localhost:5432/fides')).toThrow(
+        'AGENTD_DB_SCHEMA must be a simple Postgres identifier',
+      )
+    } finally {
+      if (previousSchema === undefined) {
+        delete process.env.AGENTD_DB_SCHEMA
+      } else {
+        process.env.AGENTD_DB_SCHEMA = previousSchema
+      }
+    }
+  })
+
   describe.skipIf(!postgresUrl && !postgresTestRequired)('postgres authority store', () => {
     if (!postgresUrl) {
       it('requires AGENTD_DATABASE_URL or DATABASE_URL when Postgres tests are mandatory', () => {
@@ -169,6 +187,55 @@ describe('agentd authority stores', () => {
       }
     }, 30_000)
 
+    it('creates and uses the configured authority schema', async () => {
+      const schema = `agentd_configured_${crypto.randomUUID().replaceAll('-', '')}`
+      const schemaIdentifier = quoteIdentifier(schema)
+      const adminSql = postgres(postgresUrl, { max: 1 })
+      const scopedUrl = postgresUrlWithSearchPath(postgresUrl, schema)
+      const scopedSql = postgres(scopedUrl, { max: 1 })
+      const previousSchema = process.env.AGENTD_DB_SCHEMA
+
+      try {
+        process.env.AGENTD_DB_SCHEMA = schema
+        await runAuthorityMigrations(scopedSql)
+
+        const schemaRows = await adminSql`
+          SELECT schema_name
+          FROM information_schema.schemata
+          WHERE schema_name = ${schema}
+        `
+        expect(schemaRows).toHaveLength(1)
+
+        const tableRows = await adminSql`
+          SELECT table_name
+          FROM information_schema.tables
+          WHERE table_schema = ${schema}
+            AND table_name IN (
+              'agentd_delegation_nonces',
+              'agentd_sessions',
+              'agentd_evidence_chains',
+              'agentd_revocations',
+              'agentd_incidents',
+              'agentd_authority_propagations',
+              'agentd_schema_migrations'
+            )
+        `
+        expect(tableRows).toHaveLength(7)
+
+        const currentSchema = await scopedSql`SELECT current_schema() AS schema`
+        expect(currentSchema[0]?.schema).toBe(schema)
+      } finally {
+        if (previousSchema === undefined) {
+          delete process.env.AGENTD_DB_SCHEMA
+        } else {
+          process.env.AGENTD_DB_SCHEMA = previousSchema
+        }
+        await scopedSql.end()
+        await adminSql.unsafe(`DROP SCHEMA IF EXISTS ${schemaIdentifier} CASCADE`)
+        await adminSql.end()
+      }
+    }, 30_000)
+
     it('fails closed when an applied migration checksum drifts', async () => {
       const schema = `agentd_migration_checksum_${crypto.randomUUID().replaceAll('-', '')}`
       const schemaIdentifier = quoteIdentifier(schema)