diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 9f9b41f83..0f8f107f4 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -283,6 +283,30 @@ jobs: test/skill-workspace-lock.test.ts test/skill-install-ledger.test.ts + # The daemon's store suites again, with `LocalStore` opened over `PostgresSyncDatabase` + # instead of `node:sqlite`, against a Testcontainers `postgres:16-alpine`. The pool runs + # that SQL for real, so a SQLite-only construct fails here rather than on a cluster. + # Needs Docker, so it gets its own runner off the critical path. + daemon-store-postgres: + name: Daemon Store (PostgreSQL) + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v6 + with: + fetch-depth: 0 + persist-credentials: false + - uses: pnpm/action-setup@v6 + with: + cache: true + - uses: actions/setup-node@v6 + with: + node-version-file: .nvmrc + - name: Install + run: pnpm install --frozen-lockfile + - name: Daemon store suites on PostgreSQL + run: pnpm --filter @agentconnect.md/daemon test:store:postgres + # control-plane integration tests against a real Postgres booted by # Testcontainers. Each Vitest worker gets an isolated database cloned from one # migrated template, so files run concurrently without cross-test TRUNCATEs. diff --git a/packages/daemon/package.json b/packages/daemon/package.json index 0818b76d4..ee8004719 100644 --- a/packages/daemon/package.json +++ b/packages/daemon/package.json @@ -33,6 +33,7 @@ "start": "node dist/index.js run", "test": "vitest run", "test:runtime-matrix": "vitest run test/acp-matrix/acp-matrix.test.ts", + "test:store:postgres": "vitest run --config vitest.postgres.config.ts", "test:unit": "vitest run", "test:watch": "vitest", "typecheck": "tsc -p tsconfig.typecheck.json" @@ -74,6 +75,7 @@ "zod": "^4.4.3" }, "devDependencies": { + "@testcontainers/postgresql": "^12.0.4", "@types/node": "^24.13.3", "@types/pg": "^8.20.0", "@types/ws": "^8.18.1", diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index f436d3b0d..2ad13150b 100644 --- a/packages/daemon/src/daemon.ts +++ b/packages/daemon/src/daemon.ts @@ -984,6 +984,10 @@ const AGENT_CALL_HOP_LIMIT_NOTICE = `Agent conversation stopped after reaching t */ const ACTIVATION_PAIRING_TTL_MS = 10 * 60 * 1000 +/** Composite-key separator for the activation rendezvous. NOT NUL: these keys and their + * transcript coordinates are stored, and the pool store is PostgreSQL, whose TEXT rejects 0x00. */ +const ACTIVATION_KEY_SEPARATOR = '\u001f' + /** * The key that makes one logical delivery admissible exactly once * (send-message-routing-rework.md §3.2). @@ -998,7 +1002,7 @@ function activationKey( platformMessageId: string, targetAgentId: string ): string { - return [platform, transportScope ?? '', platformMessageId, targetAgentId].join('\u0000') + return [platform, transportScope ?? '', platformMessageId, targetAgentId].join(ACTIVATION_KEY_SEPARATOR) } /** The platform `ts` inside a Slack `msgId` (`slack::`). The ts — not the @@ -6978,7 +6982,7 @@ export class Daemon { { agentCallDeliveryId: verified.agentCallDeliveryId, platformMessageId, - transcriptCoordinates: `${transcriptChannelKey(msg.channel, msg.transportScope)}\u0000${msg.thread ?? ''}` + transcriptCoordinates: `${transcriptChannelKey(msg.channel, msg.transportScope)}${ACTIVATION_KEY_SEPARATOR}${msg.thread ?? ''}` }, expiresAt ) @@ -8355,7 +8359,7 @@ export class Daemon { { agentCallDeliveryId: msg.trustedAgentCallDeliveryId, platformMessageId, - transcriptCoordinates: `${transcriptChannelKey(normalized.channel, normalized.transportScope)}\u0000${normalized.thread ?? ''}` + transcriptCoordinates: `${transcriptChannelKey(normalized.channel, normalized.transportScope)}${ACTIVATION_KEY_SEPARATOR}${normalized.thread ?? ''}` }, this.clock.now() + ACTIVATION_PAIRING_TTL_MS ) diff --git a/packages/daemon/test/daemon-agent-mention-routing.test.ts b/packages/daemon/test/daemon-agent-mention-routing.test.ts index fea6eb8c0..ec3f5f5db 100644 --- a/packages/daemon/test/daemon-agent-mention-routing.test.ts +++ b/packages/daemon/test/daemon-agent-mention-routing.test.ts @@ -189,7 +189,7 @@ const route = (daemon: Daemon, msg: unknown, on: string[] = ['int-bot-b']) => (d function pairingKey(daemon: Daemon, targetAgentId = 'bot-b'): string { const integrationId = (daemon as any).resolveCpAgent(targetAgentId, 'slack')?.integrationId const scope = integrationId ? (daemon as any).transportScopeForIntegrationIds([integrationId]) : undefined - return ['slack', scope ?? '', '1720000000.000200', targetAgentId].join('\u0000') + return ['slack', scope ?? '', '1720000000.000200', targetAgentId].join('\u001f') } describe('agent-authored platform mentions (send-message-routing-rework.md §6)', () => { diff --git a/packages/daemon/test/local-store-sql-portability.test.ts b/packages/daemon/test/local-store-sql-portability.test.ts new file mode 100644 index 000000000..c9655890d --- /dev/null +++ b/packages/daemon/test/local-store-sql-portability.test.ts @@ -0,0 +1,118 @@ +/** + * The daemon pool runs every `local-store.ts` statement through `PostgresSyncDatabase`, + * so a SQLite-only construct there is a production fault on the pool, not a style issue. + * `store-postgres` catches one the moment a suite covers the statement; this check is the + * cheap half — it reads the SQL text itself, so an uncovered statement still fails fast. + * + * Only constructs the pool worker does NOT rewrite are listed. `INSERT OR IGNORE`, + * `BEGIN IMMEDIATE`, `INTEGER PRIMARY KEY AUTOINCREMENT`, `PRAGMA user_version`, + * `sqlite_master`, `LIMIT -1 OFFSET` and `length(CAST(x AS BLOB))` are translated in + * `postgres-store-worker.js#rewrite` and stay legal. Extend that list only by making a + * construct portable in the SQL — never by teaching the worker another function name. + */ +import { readFileSync } from 'node:fs' +import { fileURLToPath } from 'node:url' +import { describe, expect, it } from 'vitest' +import * as ts from 'typescript' + +const storeSource = fileURLToPath(new URL('../src/store/local-store.ts', import.meta.url)) + +/** A statement may opt out with this marker plus the reason, inside the SQL itself. */ +const ALLOW_MARKER = '-- pg-portable-exempt:' + +const SQL_SHAPED = + /\b(SELECT|INSERT\s+INTO|INSERT\s+OR|UPDATE\s+|DELETE\s+FROM|CREATE\s+(TABLE|INDEX)|ALTER\s+TABLE)\b/i + +interface Fragment { + line: number + text: string +} + +/** Every string/template literal in the file, with `${…}` holes cooked to a bind placeholder. */ +function sqlFragments(): Fragment[] { + const source = readFileSync(storeSource, 'utf8') + const file = ts.createSourceFile(storeSource, source, ts.ScriptTarget.ESNext, true) + const fragments: Fragment[] = [] + const push = (node: ts.Node, text: string): void => { + if (!SQL_SHAPED.test(text)) return + fragments.push({ line: file.getLineAndCharacterOfPosition(node.getStart(file)).line + 1, text }) + } + const visit = (node: ts.Node): void => { + if (ts.isStringLiteral(node) || ts.isNoSubstitutionTemplateLiteral(node)) push(node, node.text) + else if (ts.isTemplateExpression(node)) + push(node, [node.head.text, ...node.templateSpans.map((span) => span.literal.text)].join(' ? ')) + ts.forEachChild(node, visit) + } + visit(file) + return fragments +} + +/** `MAX(a, b)` is a scalar in SQLite and an aggregate arity error in PostgreSQL — see #1068. */ +function scalarMinMax(sql: string): boolean { + for (const match of sql.matchAll(/\b(MAX|MIN)\s*\(/gi)) { + let depth = 1 + for (let i = match.index + match[0].length; i < sql.length && depth > 0; i += 1) { + const char = sql[i] + if (char === '(') depth += 1 + else if (char === ')') depth -= 1 + else if (char === ',' && depth === 1) return true + } + } + return false +} + +const CHECKS: Array<{ name: string; portable: string; hit: (sql: string) => boolean }> = [ + { name: 'two-argument MAX/MIN', portable: 'CASE, or GREATEST/LEAST', hit: scalarMinMax }, + { name: 'IFNULL', portable: 'COALESCE', hit: (sql) => /\bIFNULL\s*\(/i.test(sql) }, + { name: 'IIF', portable: 'CASE', hit: (sql) => /\bIIF\s*\(/i.test(sql) }, + { + name: 'datetime/strftime/julianday/unixepoch', + portable: 'store epoch numbers, as the schema already does', + hit: (sql) => /\b(datetime|strftime|julianday|unixepoch)\s*\(/i.test(sql) + }, + { + name: 'INSERT OR REPLACE/ABORT/FAIL', + portable: 'ON CONFLICT … DO UPDATE', + hit: (sql) => /\bINSERT\s+OR\s+(REPLACE|ABORT|FAIL)\b/i.test(sql) + }, + { name: 'GROUP_CONCAT', portable: 'string_agg', hit: (sql) => /\bGROUP_CONCAT\s*\(/i.test(sql) }, + { name: 'printf', portable: 'format, or build the string in TypeScript', hit: (sql) => /\bprintf\s*\(/i.test(sql) }, + { name: 'TYPEOF', portable: 'a typed column', hit: (sql) => /\bTYPEOF\s*\(/i.test(sql) }, + { name: '|| concatenation', portable: 'CONCAT, or concatenate in TypeScript', hit: (sql) => sql.includes('||') }, + { name: 'comma LIMIT offset', portable: 'LIMIT … OFFSET …', hit: (sql) => /\bLIMIT\s+[@?$:]?\w+\s*,/i.test(sql) } +] + +describe('local-store SQL portability', () => { + it('sees the SQL it is supposed to police', () => { + // A parser regression that stopped finding statements would make every check below vacuous. + expect(sqlFragments().length).toBeGreaterThan(100) + }) + + it('recognizes each construct it claims to police', () => { + // Without this the whole check could rot into a set of patterns that match nothing. + const samples: Record = { + 'two-argument MAX/MIN': 'UPDATE t SET a = MAX(a, @b) WHERE k = @k', + IFNULL: 'SELECT IFNULL(a, 0) AS a FROM t', + IIF: 'SELECT IIF(a > 0, 1, 0) AS a FROM t', + 'datetime/strftime/julianday/unixepoch': "SELECT * FROM t WHERE at < datetime('now')", + 'INSERT OR REPLACE/ABORT/FAIL': 'INSERT OR REPLACE INTO t (k) VALUES (@k)', + GROUP_CONCAT: 'SELECT GROUP_CONCAT(a) AS a FROM t', + printf: "SELECT printf('%s', a) AS a FROM t", + TYPEOF: 'SELECT TYPEOF(a) AS kind FROM t', + '|| concatenation': "SELECT a || '-' || b AS k FROM t", + 'comma LIMIT offset': 'SELECT * FROM t LIMIT 10, 20' + } + for (const check of CHECKS) expect([check.name, check.hit(samples[check.name] ?? '')]).toEqual([check.name, true]) + }) + + it('uses no SQLite-only construct the pool store cannot run', () => { + const offences = sqlFragments().flatMap(({ line, text }) => + text.includes(ALLOW_MARKER) + ? [] + : CHECKS.filter((check) => check.hit(text)).map( + (check) => `local-store.ts:${line} uses ${check.name} — write ${check.portable} instead` + ) + ) + expect(offences).toEqual([]) + }) +}) diff --git a/packages/daemon/test/local-store.test.ts b/packages/daemon/test/local-store.test.ts index 063dfef67..29da770ce 100644 --- a/packages/daemon/test/local-store.test.ts +++ b/packages/daemon/test/local-store.test.ts @@ -4,16 +4,31 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { DatabaseSync } from 'node:sqlite' import { LocalStore, sessionKey, type StoreDatabase } from '../src/store/local-store.js' +import { openPostgresLocalStore, usingPostgresStore } from './store-postgres/backend.js' + +/** True in the `store-postgres` project, where every store below is the real pool store. */ +const pg = usingPostgresStore() function store(): LocalStore { + if (pg) return openPostgresLocalStore() return new LocalStore(join(mkdtempSync(join(tmpdir(), 'ac-db-')), 'local.sqlite')) } +/** A second handle on the same durable store — a daemon restart, not a new store. */ +function reopen(path: string): LocalStore { + return pg ? openPostgresLocalStore() : new LocalStore(path) +} + /** Every agent a shared-store test names belongs to one org unless the test says otherwise. */ const oneOrg = () => 'org-1' /** Two pool members over ONE database — what the shared Postgres schema is during a rollout. */ function sharedMembers(first: string, second: string): [LocalStore, LocalStore] { + if (pg) + return [ + openPostgresLocalStore({ shared: true, ownerId: first, orgForAgent: oneOrg }), + openPostgresLocalStore({ shared: true, ownerId: second, orgForAgent: oneOrg }) + ] const database = new DatabaseSync(':memory:') as unknown as StoreDatabase return [ new LocalStore({ database, shared: true, ownerId: first, orgForAgent: oneOrg }), @@ -44,7 +59,7 @@ const dropTranscriptOrg = (db: DatabaseSync): void => { `) } -describe('LocalStore schema versioning', () => { +describe.skipIf(pg)('LocalStore schema versioning', () => { const userVersion = (path: string): number => { const db = new DatabaseSync(path) const v = (db.prepare('PRAGMA user_version').get() as { user_version: number }).user_version @@ -121,7 +136,7 @@ describe('LocalStore schema versioning', () => { expect(userVersion(path)).toBe(11) }) - it('never persists the CP routing map on a shared store, and still does on an owned one', () => { + it.skipIf(pg)('never persists the CP routing map on a shared store, and still does on an owned one', () => { // One row and many members: each member's save used to erase every other member's, and each // boot hydrated whichever map was written last — a foreign `routingEpoch` with it. A shared // member now writes nothing and reads nothing, so it starts from an empty map at epoch 0. @@ -317,7 +332,7 @@ describe('LocalStore', () => { it('persists only explicitly pending session snapshots and fences stale ACKs', () => { const path = join(mkdtempSync(join(tmpdir(), 'ac-session-outbox-')), 'local.sqlite') - const first = new LocalStore(path) + const first = reopen(path) first.upsertSession({ key: sessionKey('slack', 'C1', '100.1', 'bot-a'), agentId: 'bot-a', @@ -337,7 +352,7 @@ describe('LocalStore', () => { expect(first.saveSessionMetadataSnapshot('bot-a', 'acp-1', '{"phase":"start"}', true, 2)).toBe(1) first.close() - const restored = new LocalStore(path) + const restored = reopen(path) expect(restored.nextSessionMetadataSnapshot()).toMatchObject({ agentId: 'bot-a', sessionId: 'acp-1', @@ -355,7 +370,7 @@ describe('LocalStore', () => { expect(restored.nextSessionMetadataAttemptAt()).toBe(100) restored.close() - const deferred = new LocalStore(path) + const deferred = reopen(path) expect(deferred.pendingSessionMetadataSnapshot('bot-a', 'acp-1')).toMatchObject({ failedAttempts: 1, nextAttemptAt: 100 @@ -484,7 +499,7 @@ describe('LocalStore', () => { it('stores a bounded editor approval history and expires live requests after restart', () => { const path = join(mkdtempSync(join(tmpdir(), 'ac-permission-')), 'local.sqlite') - const s = new LocalStore(path) + const s = reopen(path) s.createPermissionRequest({ id: 'request-1', agentId: 'bot-a', @@ -516,7 +531,7 @@ describe('LocalStore', () => { ]) s.close() - const reopened = new LocalStore(path) + const reopened = reopen(path) expect(reopened.listPermissionRequests('bot-a')).toMatchObject([ { id: 'request-2', status: 'allowed', resolvedAt: 250 }, { id: 'request-1', status: 'expired' } @@ -599,7 +614,7 @@ describe('LocalStore', () => { s.close() }) - it('keeps an increment a concurrent writer would otherwise have erased', () => { + it.skipIf(pg)('keeps an increment a concurrent writer would otherwise have erased', () => { // The pool shape: two members touch one session across a handover. `usage` is one JSON blob, so // a plain read-merge-write silently drops whichever writer commits first. The compare-and-set // notices and re-merges instead. @@ -1100,7 +1115,7 @@ describe('LocalStore session lifecycle (§7.3/#111/#118)', () => { it('setSessionMuted persists a cold !stop tombstone across reopen and later session creation', () => { const path = join(mkdtempSync(join(tmpdir(), 'ac-mute-')), 'local.sqlite') - let s = new LocalStore(path) + let s = reopen(path) const key = sessionKey('slack', 'C1', 'T1', 'bot-a') expect(s.getSession(key)).toBeUndefined() expect(s.isSessionMuted(key)).toBe(false) @@ -1110,7 +1125,7 @@ describe('LocalStore session lifecycle (§7.3/#111/#118)', () => { s.close() // A daemon restart before SessionManager creates the row must retain the mute. - s = new LocalStore(path) + s = reopen(path) expect(s.isSessionMuted(key)).toBe(true) // Creating/upserting the actual session mirrors but never overwrites the tombstone. seed(s, key, 'bot-a', 'idle', 100) @@ -1121,7 +1136,7 @@ describe('LocalStore session lifecycle (§7.3/#111/#118)', () => { expect(s.isSessionMuted(key)).toBe(false) s.close() - const reopened = new LocalStore(path) + const reopened = reopen(path) expect(reopened.isSessionMuted(key)).toBe(false) expect(reopened.getSession(key)?.muted).toBe(0) reopened.close() @@ -1784,7 +1799,9 @@ describe('LocalStore recovery scope on a shared store (daemon pool)', () => { const sharedPath = (): string => join(mkdtempSync(join(tmpdir(), 'ac-pool-')), 'shared.sqlite') const member = (path: string, ownerId: string): LocalStore => - new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId, orgForAgent: oneOrg }) + pg + ? openPostgresLocalStore({ shared: true, ownerId, orgForAgent: oneOrg }) + : new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId, orgForAgent: oneOrg }) it("a starting member leaves a peer's live grant and running dream untouched", () => { const path = sharedPath() @@ -1943,7 +1960,7 @@ describe('LocalStore recovery scope on a shared store (daemon pool)', () => { }) describe('LocalStore activation rendezvous (send-message-routing-rework.md §3.2/§8.6)', () => { - const KEY = ['slack', 'scope-1', '1720000000.000100', 'agent-target'].join('\u0000') + const KEY = ['slack', 'scope-1', '1720000000.000100', 'agent-target'].join('\u001f') const ENVELOPE = JSON.stringify({ callFrom: 'agent-author', hopCount: 3 }) it('admits an internal-wake-first pairing exactly once and replays the same child', () => { @@ -2079,8 +2096,8 @@ describe('LocalStore activation rendezvous (send-message-routing-rework.md §3.2 // time; admitting both would strand a delivery that never persisted. The inbox row is // the only evidence that distinguishes them. const s = store() - const durable = ['slack', 'scope-1', 'ts-durable', 'agent-target'].join('\u0000') - const lost = ['slack', 'scope-1', 'ts-lost', 'agent-target'].join('\u0000') + const durable = ['slack', 'scope-1', 'ts-durable', 'agent-target'].join('\u001f') + const lost = ['slack', 'scope-1', 'ts-lost', 'agent-target'].join('\u001f') expect(s.attachActivationEnvelope(durable, ENVELOPE, 1000, 'delivery-durable').dispatch).toBe(true) expect(s.attachActivationEnvelope(lost, ENVELOPE, 1000, 'delivery-lost').dispatch).toBe(true) // Only the first one's turn actually reached the durable queue before the crash. @@ -2114,8 +2131,8 @@ describe('LocalStore activation rendezvous (send-message-routing-rework.md §3.2 // §3.2: one channel-root post can address several agents; each must be admitted once, // and one target's admission must not consume another's. const s = store() - const a = ['slack', 'scope-1', 'ts-1', 'agent-a'].join('\u0000') - const b = ['slack', 'scope-1', 'ts-1', 'agent-b'].join('\u0000') + const a = ['slack', 'scope-1', 'ts-1', 'agent-a'].join('\u001f') + const b = ['slack', 'scope-1', 'ts-1', 'agent-b'].join('\u001f') expect(s.attachActivationEnvelope(a, ENVELOPE, 1000).dispatch).toBe(true) expect(s.attachActivationEnvelope(b, ENVELOPE, 1000).dispatch).toBe(true) s.admitActivation(a, 'child-a') @@ -2136,10 +2153,10 @@ describe('sandbox generations', () => { it('survives a reopen, because the sandbox pod the number fences does', () => { const path = join(mkdtempSync(join(tmpdir(), 'ac-generations-')), 'local.sqlite') - const first = new LocalStore(path) + const first = reopen(path) expect(first.nextSandboxGeneration('agent-a')).toBe(1) first.close() - const reopened = new LocalStore(path) + const reopened = reopen(path) expect(reopened.nextSandboxGeneration('agent-a')).toBe(2) reopened.close() }) @@ -2238,7 +2255,7 @@ describe('transcript org fence on a shared store', () => { }) }) -describe('transcript org migration from a v10 store', () => { +describe.skipIf(pg)('transcript org migration from a v10 store', () => { const v10Store = (prefix: string): string => { const path = join(mkdtempSync(join(tmpdir(), prefix)), 'local.sqlite') new LocalStore(path).close() diff --git a/packages/daemon/test/memory-capture-outbox.test.ts b/packages/daemon/test/memory-capture-outbox.test.ts index 90a6cbc37..64a74e3ea 100644 --- a/packages/daemon/test/memory-capture-outbox.test.ts +++ b/packages/daemon/test/memory-capture-outbox.test.ts @@ -4,6 +4,7 @@ import { join } from 'node:path' import { describe, expect, it, vi } from 'vitest' import { MEMORY_PLUGIN_PROFILE, type MemoryConnectionSpec, type MemoryPluginManifest } from '@agentconnect.md/protocol' import { LocalStore, type MemoryCaptureOutboxRow } from '../src/store/local-store.js' +import { openPostgresLocalStore, usingPostgresStore } from './store-postgres/backend.js' import { MEMORY_CAPTURE_INPUT_MAX_BYTES, MEMORY_CAPTURE_OUTPUT_MAX_BYTES, @@ -16,7 +17,11 @@ import type { MemoryPluginMetrics } from '../src/memory-plugin/metrics.js' const connectionId = '11111111-1111-4111-8111-111111111111' +/** True in the `store-postgres` project, where every store below is the real pool store. */ +const pg = usingPostgresStore() + function store(path?: string): LocalStore { + if (pg) return openPostgresLocalStore() return new LocalStore(path ?? join(mkdtempSync(join(tmpdir(), 'ac-memory-outbox-')), 'local.sqlite')) } diff --git a/packages/daemon/test/store-postgres/backend.ts b/packages/daemon/test/store-postgres/backend.ts new file mode 100644 index 000000000..66a6eee9e --- /dev/null +++ b/packages/daemon/test/store-postgres/backend.ts @@ -0,0 +1,38 @@ +/** + * The seam that lets one store suite run against either backend. + * + * The SQLite run leaves this module unarmed and the suites build their own + * `node:sqlite` stores exactly as before. The `store-postgres` project's setup file + * arms it with one per-worker `PostgresSyncDatabase`, and the same suites then open + * their `LocalStore` over the real pool store — the SQL text the daemon pool runs. + */ +import { LocalStore, type OrgForAgent, type StoreDatabase } from '../../src/store/local-store.js' + +let poolStoreDatabase: StoreDatabase | undefined + +/** Called by the `store-postgres` setup file; never by a suite. */ +export function armPostgresStoreBackend(database: StoreDatabase): void { + poolStoreDatabase = database +} + +/** True only while the `store-postgres` project is driving the suites. */ +export function usingPostgresStore(): boolean { + return poolStoreDatabase !== undefined +} + +/** A suite closes its store freely, so the worker-wide connection must survive that. */ +function borrowed(database: StoreDatabase): StoreDatabase { + return { exec: (sql) => database.exec(sql), prepare: (sql) => database.prepare(sql), close: () => undefined } +} + +export interface PostgresLocalStoreOptions { + shared?: boolean + ownerId?: string + orgForAgent?: OrgForAgent +} + +/** Open a `LocalStore` over this worker's pool store. Solo by default, mirroring a local daemon. */ +export function openPostgresLocalStore(options: PostgresLocalStoreOptions = {}): LocalStore { + if (!poolStoreDatabase) throw new Error('the PostgreSQL store backend is not armed — run the store-postgres project') + return new LocalStore({ database: borrowed(poolStoreDatabase), ...options }) +} diff --git a/packages/daemon/test/store-postgres/global-setup.ts b/packages/daemon/test/store-postgres/global-setup.ts new file mode 100644 index 000000000..535728648 --- /dev/null +++ b/packages/daemon/test/store-postgres/global-setup.ts @@ -0,0 +1,57 @@ +/** + * Vitest global setup for the `store-postgres` project. + * + * Boots ONE `postgres:16-alpine` via Testcontainers and hands each Vitest pool its own + * database, so the store suites run the real `LocalStore` SQL through the real + * `PostgresSyncDatabase` worker instead of SQLite. Per-test isolation is a schema-wide + * sweep in `setup.ts`; per-worker isolation is the separate database created here. + */ +import { PostgreSqlContainer, type StartedPostgreSqlContainer } from '@testcontainers/postgresql' +import type { ProvidedContext } from 'vitest' +import { storePostgresWorkerCount } from './workers.js' + +let container: StartedPostgreSqlContainer | undefined + +/** Vitest 4 passes a `TestProject` but exports no `GlobalSetupContext`, so type what we use. */ +interface GlobalSetupContext { + provide(key: K, value: ProvidedContext[K]): void +} + +function databaseUrl(baseUrl: string, database: string): string { + const url = new URL(baseUrl) + url.pathname = `/${database}` + return url.toString() +} + +export async function setup({ provide }: GlobalSetupContext): Promise { + container = await new PostgreSqlContainer('postgres:16-alpine').start() + const base = container.getConnectionUri() + const urls: string[] = [] + for (let poolId = 1; poolId <= storePostgresWorkerCount(); poolId += 1) { + const database = `store_worker_${poolId}` + const result = await container.exec([ + 'psql', + '-v', + 'ON_ERROR_STOP=1', + '-U', + container.getUsername(), + '-d', + 'postgres', + '-c', + `CREATE DATABASE "${database}" OWNER "${container.getUsername()}"` + ]) + if (result.exitCode !== 0) throw new Error(`failed to create ${database}: ${result.output}`) + urls.push(databaseUrl(base, database)) + } + provide('storeDatabaseUrls', urls) +} + +export async function teardown(): Promise { + await container?.stop() +} + +declare module 'vitest' { + export interface ProvidedContext { + storeDatabaseUrls: string[] + } +} diff --git a/packages/daemon/test/store-postgres/setup.ts b/packages/daemon/test/store-postgres/setup.ts new file mode 100644 index 000000000..68195029b --- /dev/null +++ b/packages/daemon/test/store-postgres/setup.ts @@ -0,0 +1,44 @@ +/** + * Per-worker harness for the `store-postgres` project. + * + * One `PostgresSyncDatabase` per Vitest pool, on that pool's own database from + * `global-setup.ts`. The first `LocalStore` materializes the schema through the very + * SQLite→PostgreSQL rewrite the daemon pool uses, then the advisory lock is released so + * a suite may open its own `PostgresDataPlane` on the same database. Per-test isolation + * is a schema-wide sweep, the way the control-plane integration project does it. + */ +import { afterAll, beforeEach, inject } from 'vitest' +import { PostgresSyncDatabase } from '../../src/store/postgres-sync-database.js' +import { armPostgresStoreBackend, openPostgresLocalStore } from './backend.js' + +const poolId = Number(process.env.VITEST_POOL_ID ?? '1') +const databaseUrl = inject('storeDatabaseUrls')[poolId - 1] +if (!databaseUrl) throw new Error(`No store database provisioned for Vitest pool ${poolId}`) + +// The gated pool-store suites read this; in this project they are no longer gated out. +process.env.DATA_PLANE_TEST_DATABASE_URL = databaseUrl + +const database = new PostgresSyncDatabase({ version: 1, databaseUrl, maxConnections: 2 }) +armPostgresStoreBackend(database) +openPostgresLocalStore().close() +database.finishSchemaInitialization() + +// Empty every store table between tests and rewind the pool's revision sequence, so a +// suite sees the same blank store SQLite hands it from a fresh temporary file. +const SWEEP_SQL = `DO $sweep$ +DECLARE tables text; +BEGIN + SELECT string_agg(format('%I', c.relname), ', ') INTO tables FROM pg_class c + WHERE c.relnamespace = 'agentconnect_cloud_store'::regnamespace AND c.relkind = 'r' + AND c.relname <> '_local_store_schema_version'; + IF tables IS NOT NULL THEN EXECUTE format('TRUNCATE TABLE %s RESTART IDENTITY CASCADE', tables); END IF; + PERFORM setval('_transcript_revision_seq', 1, false); +END $sweep$;` + +beforeEach(() => { + database.exec(SWEEP_SQL) +}) + +afterAll(() => { + database.close() +}) diff --git a/packages/daemon/test/store-postgres/workers.ts b/packages/daemon/test/store-postgres/workers.ts new file mode 100644 index 000000000..40e613600 --- /dev/null +++ b/packages/daemon/test/store-postgres/workers.ts @@ -0,0 +1,12 @@ +const DEFAULT_STORE_POSTGRES_WORKERS = 2 + +/** Keep the Vitest worker count and the number of per-worker databases in lockstep. */ +export function storePostgresWorkerCount(env: NodeJS.ProcessEnv = process.env): number { + const raw = env.STORE_POSTGRES_TEST_WORKERS + if (raw === undefined) return DEFAULT_STORE_POSTGRES_WORKERS + const workers = Number(raw) + if (!Number.isSafeInteger(workers) || workers < 1) { + throw new Error(`STORE_POSTGRES_TEST_WORKERS must be a positive integer, got ${JSON.stringify(raw)}`) + } + return workers +} diff --git a/packages/daemon/test/webchat-continuation-fixture.ts b/packages/daemon/test/webchat-continuation-fixture.ts index abd9c1f3d..7f980c3bd 100644 --- a/packages/daemon/test/webchat-continuation-fixture.ts +++ b/packages/daemon/test/webchat-continuation-fixture.ts @@ -168,4 +168,4 @@ export const settle = () => new Promise((r) => setTimeout(r, 300)) /** Mirrors daemon.ts `activationKey(platform, transportScope, platformMessageId, target)`. */ export const rendezvousKey = (postId: string, targetAgentId: string): string => - ['webchat', '', postId, targetAgentId].join('\u0000') + ['webchat', '', postId, targetAgentId].join('\u001f') diff --git a/packages/daemon/vitest.postgres.config.ts b/packages/daemon/vitest.postgres.config.ts new file mode 100644 index 000000000..6790caebc --- /dev/null +++ b/packages/daemon/vitest.postgres.config.ts @@ -0,0 +1,34 @@ +import { defineConfig } from 'vitest/config' +import { storePostgresWorkerCount } from './test/store-postgres/workers.js' +import { githubActionsReporters } from '../../scripts/vitest-github-reporters.js' + +/** + * The `store-postgres` project: the daemon's store suites re-run with `LocalStore` opened + * over `PostgresSyncDatabase` instead of `node:sqlite`, against a Testcontainers + * `postgres:16-alpine`. The pool runs this SQL for real, so SQLite-only constructs + * (two-arg `MAX`/`MIN`, `IFNULL`, `datetime()`, …) fail here instead of on a cluster. + * + * A separate config file, not a second project in `vitest.config.ts`, so `vitest run` and + * every targeted `vitest run ` in CI stay Docker-free. + */ +export default defineConfig({ + test: { + name: 'store-postgres', + environment: 'node', + include: [ + 'test/local-store.test.ts', + 'test/memory-capture-outbox.test.ts', + 'test/postgres-pool-store.int.test.ts', + 'test/postgres-transcript-org.int.test.ts' + ], + globalSetup: ['./test/store-postgres/global-setup.ts'], + setupFiles: ['./test/store-postgres/setup.ts'], + maxWorkers: storePostgresWorkerCount(), + fileParallelism: true, + hookTimeout: 120_000, + // Every statement is a round trip to a Dockerized Postgres through a worker thread, + // so Vitest's 5s unit budget would time out on runner weather, not on the code. + testTimeout: 30_000, + reporters: githubActionsReporters('daemon-store-postgres.md') + } +}) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7f481684d..b31827098 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -293,7 +293,7 @@ importers: version: 0.0.67 '@larksuiteoapi/node-sdk': specifier: ^1.71.1 - version: 1.71.1 + version: 1.71.1(debug@4.3.4) '@modelcontextprotocol/client': specifier: 2.0.0 version: 2.0.0 @@ -373,6 +373,9 @@ importers: specifier: ^4.4.3 version: 4.4.3 devDependencies: + '@testcontainers/postgresql': + specifier: ^12.0.4 + version: 12.0.4 '@types/node': specifier: ^24.13.3 version: 24.13.3 @@ -10226,9 +10229,9 @@ snapshots: '@kwsites/promise-deferred@1.1.1': {} - '@larksuiteoapi/node-sdk@1.71.1': + '@larksuiteoapi/node-sdk@1.71.1(debug@4.3.4)': dependencies: - axios: 1.18.1 + axios: 1.18.1(debug@4.3.4) lodash.identity: 3.0.0 lodash.merge: 4.6.2 lodash.pickby: 4.6.0 @@ -11586,7 +11589,7 @@ snapshots: '@slack/types': 2.22.0 '@types/node': 24.13.3 '@types/retry': 0.12.0 - axios: 1.18.1 + axios: 1.18.1(debug@4.3.4) eventemitter3: 5.0.4 form-data: 4.0.6 is-electron: 2.2.2 @@ -12606,7 +12609,7 @@ snapshots: - supports-color optional: true - axios@1.18.1: + axios@1.18.1(debug@4.3.4): dependencies: follow-redirects: 1.16.0(debug@4.3.4) form-data: 4.0.6