diff --git a/.changeset/fix-audit-append-order.md b/.changeset/fix-audit-append-order.md new file mode 100644 index 0000000000..6a38acd2ad --- /dev/null +++ b/.changeset/fix-audit-append-order.md @@ -0,0 +1,5 @@ +--- +"@agent-native/core": patch +--- + +Preserve audit event append order when timestamps are equal. diff --git a/packages/core/src/audit/store.postgres.integration.spec.ts b/packages/core/src/audit/store.postgres.integration.spec.ts new file mode 100644 index 0000000000..03e150247b --- /dev/null +++ b/packages/core/src/audit/store.postgres.integration.spec.ts @@ -0,0 +1,197 @@ +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; + +import type { AuditEvent } from "./types.js"; + +const postgresUrl = process.env.AGENT_NATIVE_AUDIT_POSTGRES_URL; + +describe.skipIf(!postgresUrl)("audit store PostgreSQL append order", () => { + let dbModule: typeof import("../db/client.js"); + let ddlModule: typeof import("../db/ddl-guard.js"); + let store: typeof import("./store.js"); + + const event = (id: string, createdAt = 700): AuditEvent => ({ + id, + createdAt, + action: "update-schedule", + caller: "tool", + actorKind: "agent", + actorEmail: "alice@example.test", + orgId: null, + threadId: null, + turnId: null, + targetType: "automation", + targetId: "automation-1", + status: "success", + summary: null, + input: null, + errorCode: null, + ownerEmail: "alice@example.test", + visibility: "private", + }); + + async function resetEphemeralSchema(): Promise { + const client = dbModule.getDbExec(); + await client.execute("DROP TABLE IF EXISTS agent_audit_log CASCADE"); + await client.execute( + "DROP TABLE IF EXISTS agent_audit_append_order CASCADE", + ); + await client.execute( + "DROP FUNCTION IF EXISTS agent_audit_allocate_append_order() CASCADE", + ); + await dbModule.closeDbExec(); + ddlModule.__resetSchemaSnapshotForTests(); + store.__resetAuditInitForTests(); + } + + beforeAll(async () => { + const parsed = new URL(postgresUrl!); + expect(["127.0.0.1", "localhost", "::1"]).toContain(parsed.hostname); + expect(parsed.pathname).toBe("/audit_test"); + vi.stubEnv("DATABASE_URL", postgresUrl!); + + dbModule = await import("../db/client.js"); + ddlModule = await import("../db/ddl-guard.js"); + store = await import("./store.js"); + + const version = await dbModule.getDbExec().execute("SHOW server_version"); + expect(String(version.rows[0]?.server_version)).toMatch(/^17\./); + }); + + beforeEach(resetEphemeralSchema); + + afterAll(async () => { + if (dbModule) { + await resetEphemeralSchema(); + await dbModule.closeDbExec(); + } + vi.unstubAllEnvs(); + }); + + it("returns exact tied-timestamp pages in append order", async () => { + await store.insertAuditEvent(event("z-first")); + await store.insertAuditEvent(event("m-second")); + await store.insertAuditEvent(event("a-third")); + + const firstPage = await store.queryAuditEvents( + { userEmail: "alice@example.test" }, + { limit: 2 }, + ); + const secondPage = await store.queryAuditEvents( + { userEmail: "alice@example.test" }, + { limit: 2, offset: 2 }, + ); + + expect(firstPage.map((row) => row.id)).toEqual(["a-third", "m-second"]); + expect(secondPage.map((row) => row.id)).toEqual(["z-first"]); + const union = [...firstPage, ...secondPage]; + expect(union.map((row) => row.id)).toEqual([ + "a-third", + "m-second", + "z-first", + ]); + expect(new Set(union.map((row) => row.id)).size).toBe(3); + expect(union.every((row) => !("append_order" in row))).toBe(true); + }); + + it("backfills a legacy table and allocates for rolling old writers", async () => { + const client = dbModule.getDbExec(); + await client.execute(` + CREATE TABLE agent_audit_log ( + id TEXT PRIMARY KEY, + created_at BIGINT NOT NULL, + action TEXT NOT NULL, + caller TEXT NOT NULL, + actor_kind TEXT NOT NULL, + actor_email TEXT, + org_id TEXT, + thread_id TEXT, + turn_id TEXT, + target_type TEXT, + target_id TEXT, + status TEXT NOT NULL DEFAULT 'success', + summary TEXT, + input TEXT, + error_code TEXT, + owner_email TEXT, + visibility TEXT NOT NULL DEFAULT 'private' + ) + `); + for (const id of ["legacy-z", "legacy-m", "legacy-a"]) { + await client.execute({ + sql: `INSERT INTO agent_audit_log + (id, created_at, action, caller, actor_kind, owner_email, visibility) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + args: [ + id, + 700, + "legacy", + "tool", + "agent", + "alice@example.test", + "private", + ], + }); + } + + store.__resetAuditInitForTests(); + await store.ensureAuditTables(); + + const migrated = await client.execute( + "SELECT id, append_order FROM agent_audit_log ORDER BY append_order", + ); + expect(migrated.rows.map((row) => row.id)).toEqual([ + "legacy-z", + "legacy-m", + "legacy-a", + ]); + expect(migrated.rows.map((row) => Number(row.append_order))).toEqual([ + 1, 2, 3, + ]); + + // This is the pre-upgrade INSERT shape: it omits append_order entirely. + await client.execute({ + sql: `INSERT INTO agent_audit_log + (id, created_at, action, caller, actor_kind, owner_email, visibility) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + args: [ + "a-after-upgrade", + 700, + "legacy-writer", + "tool", + "agent", + "alice@example.test", + "private", + ], + }); + const state = await client.execute(` + SELECT COUNT(*)::INT AS total, + COUNT(append_order)::INT AS non_null, + COUNT(DISTINCT append_order)::INT AS unique_count, + MAX(append_order)::BIGINT AS max_order + FROM agent_audit_log + `); + expect(state.rows[0]).toMatchObject({ + total: 4, + non_null: 4, + unique_count: 4, + max_order: "4", + }); + const rows = await store.queryAuditEvents({ + userEmail: "alice@example.test", + }); + expect(rows.map((row) => row.id)).toEqual([ + "a-after-upgrade", + "legacy-a", + "legacy-m", + "legacy-z", + ]); + }); +}); diff --git a/packages/core/src/audit/store.spec.ts b/packages/core/src/audit/store.spec.ts index 5cbfd3bae6..89675c322b 100644 --- a/packages/core/src/audit/store.spec.ts +++ b/packages/core/src/audit/store.spec.ts @@ -13,7 +13,7 @@ const rawClient = { } const stmt = sqlite.prepare(input.sql); const args = (input.args ?? []) as unknown[]; - if (/^\s*select/i.test(input.sql)) { + if (stmt.reader) { return { rows: stmt.all(...args), rowsAffected: 0 }; } const info = stmt.run(...args); @@ -207,6 +207,102 @@ describe("audit store filters + ordering", () => { expect(limited).toHaveLength(2); }); + it("uses append order for exact tied-timestamp pages", async () => { + const createdAt = 700; + await insertAuditEvent(makeEvent({ id: "z-first", createdAt })); + await insertAuditEvent(makeEvent({ id: "m-second", createdAt })); + await insertAuditEvent(makeEvent({ id: "a-third", createdAt })); + + const firstPage = await queryAuditEvents( + { userEmail: "alice@x.com" }, + { limit: 2 }, + ); + const secondPage = await queryAuditEvents( + { userEmail: "alice@x.com" }, + { limit: 2, offset: 2 }, + ); + + expect(firstPage.map((row) => row.id)).toEqual(["a-third", "m-second"]); + expect(secondPage.map((row) => row.id)).toEqual(["z-first"]); + expect([...firstPage, ...secondPage].map((row) => row.id)).toEqual([ + "a-third", + "m-second", + "z-first", + ]); + expect( + new Set([...firstPage, ...secondPage].map((row) => row.id)).size, + ).toBe(3); + }); + + it("initializes append order for a legacy SQLite table", async () => { + sqlite.close(); + sqlite = new Database(":memory:"); + sqlite.exec(` + CREATE TABLE agent_audit_log ( + id TEXT PRIMARY KEY, + created_at INTEGER NOT NULL, + action TEXT NOT NULL, + caller TEXT NOT NULL, + actor_kind TEXT NOT NULL, + actor_email TEXT, + org_id TEXT, + thread_id TEXT, + turn_id TEXT, + target_type TEXT, + target_id TEXT, + status TEXT NOT NULL DEFAULT 'success', + summary TEXT, + input TEXT, + error_code TEXT, + owner_email TEXT, + visibility TEXT NOT NULL DEFAULT 'private' + ); + INSERT INTO agent_audit_log + (id, created_at, action, caller, actor_kind, owner_email, visibility) + VALUES + ('legacy-z', 700, 'legacy', 'tool', 'agent', 'alice@x.com', 'private'), + ('legacy-m', 700, 'legacy', 'tool', 'agent', 'alice@x.com', 'private'), + ('legacy-a', 700, 'legacy', 'tool', 'agent', 'alice@x.com', 'private'); + `); + __resetAuditInitForTests(); + + await ensureAuditTables(); + const migrated = sqlite + .prepare( + "SELECT id, append_order FROM agent_audit_log ORDER BY append_order", + ) + .all() as Array<{ id: string; append_order: number }>; + expect(migrated).toEqual([ + { id: "legacy-z", append_order: 1 }, + { id: "legacy-m", append_order: 2 }, + { id: "legacy-a", append_order: 3 }, + ]); + + await insertAuditEvent( + makeEvent({ id: "a-after-upgrade", createdAt: 700 }), + ); + const rows = await queryAuditEvents({ userEmail: "alice@x.com" }); + expect(rows.map((row) => row.id)).toEqual([ + "a-after-upgrade", + "legacy-a", + "legacy-m", + "legacy-z", + ]); + const nullOrDuplicate = sqlite + .prepare( + `SELECT COUNT(*) AS total, + COUNT(append_order) AS non_null, + COUNT(DISTINCT append_order) AS unique_count + FROM agent_audit_log`, + ) + .get() as { total: number; non_null: number; unique_count: number }; + expect(nullOrDuplicate).toEqual({ + total: 4, + non_null: 4, + unique_count: 4, + }); + }); + it("filters by sinceMs", async () => { await insertAuditEvent(makeEvent({ createdAt: 100 })); await insertAuditEvent(makeEvent({ createdAt: 500 })); diff --git a/packages/core/src/audit/store.ts b/packages/core/src/audit/store.ts index baccc30928..51f2deb442 100644 --- a/packages/core/src/audit/store.ts +++ b/packages/core/src/audit/store.ts @@ -7,11 +7,14 @@ * `agent_audit_log`; reads are scoped to the caller's identity in SQL (no * shares table — audit rows are never individually shared). */ -import { getDbExec, intType, isPostgres } from "../db/client.js"; +import { getDbExec, intType, isPostgres, type DbExec } from "../db/client.js"; import { ensureColumnExists, ensureTableExists, ensureIndexExists, + pgColumnExists, + pgIndexExists, + pgTableExists, } from "../db/ddl-guard.js"; import type { AuditEvent, @@ -21,6 +24,221 @@ import type { let _initPromise: Promise | undefined; +const APPEND_ORDER_TABLE = "agent_audit_append_order"; +const APPEND_ORDER_TRIGGER = "agent_audit_assign_append_order"; +const APPEND_ORDER_UNIQUE_INDEX = "idx_audit_append_order_unique"; +const APPEND_ORDER_QUERY_INDEX = "idx_audit_created_append_order"; + +async function postgresAppendOrderReady(client: DbExec): Promise { + const probes = await Promise.all([ + pgColumnExists("agent_audit_log", "append_order", client), + pgTableExists(APPEND_ORDER_TABLE, client, true), + pgIndexExists(APPEND_ORDER_UNIQUE_INDEX, client, true), + pgIndexExists(APPEND_ORDER_QUERY_INDEX, client, true), + ]); + if (probes.some((value) => value === undefined)) { + throw new Error( + "Could not probe the audit append-order schema; refusing to issue DDL", + ); + } + if (probes.some((value) => value !== true)) return false; + + const metadata = await client.execute(` + SELECT + EXISTS ( + SELECT 1 + FROM pg_trigger + WHERE tgname = '${APPEND_ORDER_TRIGGER}' + AND tgrelid = 'agent_audit_log'::regclass + AND NOT tgisinternal + AND tgenabled <> 'D' + ) AS trigger_ready, + EXISTS ( + SELECT 1 + FROM pg_attribute + WHERE attrelid = 'agent_audit_log'::regclass + AND attname = 'append_order' + AND attnotnull + AND NOT attisdropped + ) AS column_not_null + `); + const allocator = await client.execute( + `SELECT value FROM ${APPEND_ORDER_TABLE} WHERE id = 1`, + ); + const state = metadata.rows[0]; + return ( + state?.trigger_ready === true && + state?.column_not_null === true && + allocator.rows.length === 1 + ); +} + +async function ensurePostgresAppendOrder(client: DbExec): Promise { + if (await postgresAppendOrderReady(client)) return; + if (!client.transaction) { + throw new Error( + "PostgreSQL audit append-order initialization requires a database transaction", + ); + } + + await client.transaction(async (tx) => { + await tx.execute("SET LOCAL lock_timeout = '3s'"); + await tx.execute("SET LOCAL idle_in_transaction_session_timeout = '30s'"); + // The lock closes the add-column -> trigger window for rolling old writers. + await tx.execute("LOCK TABLE agent_audit_log IN SHARE ROW EXCLUSIVE MODE"); + + await ensureTableExists( + APPEND_ORDER_TABLE, + `CREATE TABLE IF NOT EXISTS ${APPEND_ORDER_TABLE} ( + id INTEGER PRIMARY KEY CHECK (id = 1), + value BIGINT NOT NULL + )`, + { injectedClient: tx, dialectIsPostgres: true }, + ); + await ensureColumnExists( + "agent_audit_log", + "append_order", + "ALTER TABLE agent_audit_log ADD COLUMN IF NOT EXISTS append_order BIGINT", + { injectedClient: tx }, + ); + + await tx.execute(` + CREATE OR REPLACE FUNCTION agent_audit_allocate_append_order() + RETURNS trigger + LANGUAGE plpgsql + AS $audit_append_order$ + DECLARE + allocated BIGINT; + BEGIN + UPDATE ${APPEND_ORDER_TABLE} + SET value = value + 1 + WHERE id = 1 + RETURNING value INTO allocated; + IF allocated IS NULL THEN + RAISE EXCEPTION 'audit append-order allocator is not initialized'; + END IF; + NEW.append_order := allocated; + RETURN NEW; + END; + $audit_append_order$ + `); + await tx.execute(` + CREATE OR REPLACE TRIGGER ${APPEND_ORDER_TRIGGER} + BEFORE INSERT ON agent_audit_log + FOR EACH ROW + EXECUTE FUNCTION agent_audit_allocate_append_order() + `); + + // PostgreSQL cannot recover true append order for historical ties. This + // migration rank is stable and database-owned; the live guarantee starts + // with rows inserted after this transaction commits. + await tx.execute(` + WITH base AS ( + SELECT COALESCE(MAX(append_order), 0) AS value + FROM agent_audit_log + ), ranked AS ( + SELECT agent_audit_log.ctid AS row_id, + base.value + ROW_NUMBER() OVER ( + ORDER BY agent_audit_log.created_at, agent_audit_log.ctid + ) AS value + FROM agent_audit_log + CROSS JOIN base + WHERE append_order IS NULL + ) + UPDATE agent_audit_log AS audit + SET append_order = ranked.value + FROM ranked + WHERE audit.ctid = ranked.row_id + `); + await tx.execute(` + INSERT INTO ${APPEND_ORDER_TABLE} (id, value) + SELECT 1, COALESCE(MAX(append_order), 0) FROM agent_audit_log + ON CONFLICT (id) DO UPDATE + SET value = GREATEST(${APPEND_ORDER_TABLE}.value, excluded.value) + `); + await tx.execute( + "ALTER TABLE agent_audit_log ALTER COLUMN append_order SET NOT NULL", + ); + await ensureIndexExists( + APPEND_ORDER_UNIQUE_INDEX, + `CREATE UNIQUE INDEX IF NOT EXISTS ${APPEND_ORDER_UNIQUE_INDEX} ON agent_audit_log (append_order)`, + { injectedClient: tx, dialectIsPostgres: true }, + ); + await ensureIndexExists( + APPEND_ORDER_QUERY_INDEX, + `CREATE INDEX IF NOT EXISTS ${APPEND_ORDER_QUERY_INDEX} ON agent_audit_log (created_at DESC, append_order DESC)`, + { injectedClient: tx, dialectIsPostgres: true }, + ); + }); +} + +async function ensureSqliteAppendOrder(client: DbExec): Promise { + const initialize = async (tx: DbExec) => { + await tx.execute(`CREATE TABLE IF NOT EXISTS ${APPEND_ORDER_TABLE} ( + id INTEGER PRIMARY KEY CHECK (id = 1), + value INTEGER NOT NULL + )`); + const columns = await tx.execute({ + sql: "PRAGMA table_info(agent_audit_log)", + }); + if (!columns.rows.some((row) => row.name === "append_order")) { + await tx.execute( + "ALTER TABLE agent_audit_log ADD COLUMN append_order INTEGER", + ); + } + + // A rowid-ranked backfill preserves SQLite's database-owned legacy order. + await tx.execute(` + WITH base AS ( + SELECT COALESCE(MAX(append_order), 0) AS value + FROM agent_audit_log + ), ranked AS MATERIALIZED ( + SELECT rowid AS row_id, + base.value + ROW_NUMBER() OVER (ORDER BY rowid) AS value + FROM agent_audit_log + CROSS JOIN base + WHERE append_order IS NULL + ) + UPDATE agent_audit_log + SET append_order = ( + SELECT ranked.value FROM ranked WHERE ranked.row_id = agent_audit_log.rowid + ) + WHERE append_order IS NULL + `); + await tx.execute(` + INSERT INTO ${APPEND_ORDER_TABLE} (id, value) + VALUES (1, (SELECT COALESCE(MAX(append_order), 0) FROM agent_audit_log)) + ON CONFLICT (id) DO UPDATE + SET value = MAX(${APPEND_ORDER_TABLE}.value, excluded.value) + `); + await tx.execute(` + CREATE TRIGGER IF NOT EXISTS ${APPEND_ORDER_TRIGGER} + AFTER INSERT ON agent_audit_log + FOR EACH ROW WHEN NEW.append_order IS NULL + BEGIN + UPDATE ${APPEND_ORDER_TABLE} SET value = value + 1 WHERE id = 1; + SELECT CASE WHEN changes() = 0 + THEN RAISE(ABORT, 'audit append-order allocator is not initialized') END; + UPDATE agent_audit_log + SET append_order = (SELECT value FROM ${APPEND_ORDER_TABLE} WHERE id = 1) + WHERE rowid = NEW.rowid; + END + `); + await tx.execute( + `CREATE UNIQUE INDEX IF NOT EXISTS ${APPEND_ORDER_UNIQUE_INDEX} ON agent_audit_log (append_order)`, + ); + await tx.execute( + `CREATE INDEX IF NOT EXISTS ${APPEND_ORDER_QUERY_INDEX} ON agent_audit_log (created_at DESC, append_order DESC)`, + ); + }; + + if (client.transaction) { + await client.transaction(initialize); + } else { + await initialize(client); + } +} + export async function ensureAuditTables(): Promise { if (!_initPromise) { _initPromise = (async () => { @@ -29,6 +247,7 @@ export async function ensureAuditTables(): Promise { CREATE TABLE IF NOT EXISTS agent_audit_log ( id TEXT PRIMARY KEY, created_at ${intType()} NOT NULL, + append_order ${intType()}, action TEXT NOT NULL, caller TEXT NOT NULL, actor_kind TEXT NOT NULL, @@ -104,6 +323,7 @@ export async function ensureAuditTables(): Promise { "idx_audit_created", `CREATE INDEX IF NOT EXISTS idx_audit_created ON agent_audit_log (created_at)`, ); + await ensurePostgresAppendOrder(client); return; } @@ -131,6 +351,7 @@ export async function ensureAuditTables(): Promise { // Index creation is best-effort; a racing boot may have created it. } } + await ensureSqliteAppendOrder(client); })().catch((err) => { // Allow a later call to retry if the first init failed. _initPromise = undefined; @@ -306,7 +527,7 @@ export async function queryAuditEvents( const result = await client.execute({ sql: `SELECT ${LIST_COLUMNS} FROM agent_audit_log WHERE ${where.join(" AND ")} - ORDER BY created_at DESC + ORDER BY created_at DESC, append_order DESC LIMIT ? OFFSET ?`, args: [...args, limit, offset], });