diff --git a/hub/src/store/eventStore.ts b/hub/src/store/eventStore.ts index 74a10076ff..762dffe59f 100644 --- a/hub/src/store/eventStore.ts +++ b/hub/src/store/eventStore.ts @@ -2,6 +2,7 @@ import type { Database } from 'bun:sqlite' import { countSystemEvents, getSystemEventById, + getSystemEventByDedupeKey, getSystemEventByIdempotencyKey, insertEventLink, insertSystemEvent, @@ -39,6 +40,10 @@ export class EventStore { return getSystemEventByIdempotencyKey(this.db, idempotencyKey) } + getByDedupeKey(dedupeKey: string, namespace = 'default'): StoredSystemEvent | null { + return getSystemEventByDedupeKey(this.db, dedupeKey, namespace) + } + count(): number { return countSystemEvents(this.db) } diff --git a/hub/src/store/events.schema.test.ts b/hub/src/store/events.schema.test.ts new file mode 100644 index 0000000000..2b006baa48 --- /dev/null +++ b/hub/src/store/events.schema.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'bun:test' +import { Database } from 'bun:sqlite' +import { ensureOverseerEventsSchema } from './events' + +describe('ensureOverseerEventsSchema', () => { + it('migrates legacy events table that predates namespace column', () => { + const db = new Database(':memory:') + db.exec(` + CREATE TABLE events ( + id INTEGER PRIMARY KEY, + ts INTEGER NOT NULL, + source_kind TEXT NOT NULL, + source_ref TEXT, + sink_kind TEXT, + sink_ref TEXT, + event_type TEXT NOT NULL, + attention_candidate INTEGER NOT NULL DEFAULT 0, + operator_action_required INTEGER NOT NULL DEFAULT 0, + risk_detected INTEGER NOT NULL DEFAULT 0, + summary TEXT NOT NULL, + payload_json TEXT, + artifact_refs TEXT, + tags TEXT, + related_session_id TEXT, + related_event_id INTEGER, + dedupe_key TEXT, + expires_at INTEGER, + provenance TEXT, + idempotency_key TEXT, + confidence REAL, + severity INTEGER + ); + CREATE UNIQUE INDEX idx_events_dedupe_key ON events(dedupe_key) WHERE dedupe_key IS NOT NULL; + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + namespace TEXT NOT NULL DEFAULT 'default' + ); + INSERT INTO sessions (id, namespace) VALUES ('legacy-foreign-sess', 'other-ns'); + INSERT INTO events ( + ts, source_kind, event_type, attention_candidate, summary, + related_session_id, dedupe_key, provenance + ) VALUES ( + 1, 'channel', 'blocked', 1, 'legacy foreign event', + 'legacy-foreign-sess', 'contrib:tiann/hapi#1:blocked', 'test' + ); + `) + + expect(() => ensureOverseerEventsSchema(db)).not.toThrow() + + const row = db.prepare( + `SELECT namespace FROM events WHERE dedupe_key = 'contrib:tiann/hapi#1:blocked'` + ).get() as { namespace: string } + expect(row.namespace).toBe('other-ns') + + const columns = db.prepare('PRAGMA table_info(events)').all() as Array<{ name: string }> + expect(columns.some((column) => column.name === 'namespace')).toBe(true) + + const index = db.prepare( + `SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_events_namespace_dedupe_key'` + ).get() as { name: string } | null + expect(index?.name).toBe('idx_events_namespace_dedupe_key') + }) +}) diff --git a/hub/src/store/events.ts b/hub/src/store/events.ts index c9876fb896..2e6e530c8a 100644 --- a/hub/src/store/events.ts +++ b/hub/src/store/events.ts @@ -24,6 +24,7 @@ export type InsertSystemEventInput = { idempotencyKey?: string | null confidence?: number | null severity?: number | null + namespace?: string | null } export type StoredSystemEvent = { @@ -49,6 +50,7 @@ export type StoredSystemEvent = { idempotencyKey: string | null confidence: number | null severity: number | null + namespace: string } export type ListSystemEventsOptions = { @@ -96,6 +98,7 @@ type SystemEventRow = { idempotency_key: string | null confidence: number | null severity: number | null + namespace: string } function mapRow(row: SystemEventRow): StoredSystemEvent { @@ -121,7 +124,8 @@ function mapRow(row: SystemEventRow): StoredSystemEvent { provenance: row.provenance, idempotencyKey: row.idempotency_key, confidence: row.confidence, - severity: row.severity + severity: row.severity, + namespace: row.namespace } } @@ -152,12 +156,17 @@ export function repointSessionEvents(db: Database, fromSessionId: string, toSess } export function insertSystemEvent(db: Database, input: InsertSystemEventInput): StoredSystemEvent | null { + const namespace = input.namespace ?? 'default' if (input.idempotencyKey) { - const existing = db.prepare( - 'SELECT id FROM events WHERE idempotency_key = ? LIMIT 1' - ).get(input.idempotencyKey) as { id: number } | undefined + const existing = getSystemEventByIdempotencyKey(db, input.idempotencyKey) if (existing) { - return getSystemEventById(db, existing.id) + return existing + } + } + if (input.dedupeKey) { + const existing = getSystemEventByDedupeKey(db, input.dedupeKey, namespace) + if (existing) { + return existing } } @@ -167,13 +176,13 @@ export function insertSystemEvent(db: Database, input: InsertSystemEventInput): event_type, attention_candidate, operator_action_required, risk_detected, summary, payload_json, artifact_refs, tags, related_session_id, related_event_id, dedupe_key, expires_at, - provenance, idempotency_key, confidence, severity + provenance, idempotency_key, confidence, severity, namespace ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, - ?, ?, ?, ? + ?, ?, ?, ?, ? ) `) @@ -198,7 +207,8 @@ export function insertSystemEvent(db: Database, input: InsertSystemEventInput): input.provenance ?? null, input.idempotencyKey ?? null, input.confidence ?? null, - input.severity ?? null + input.severity ?? null, + namespace ) const id = Number(result.lastInsertRowid) @@ -251,6 +261,17 @@ export function getSystemEventByIdempotencyKey(db: Database, idempotencyKey: str return row ? mapRow(row) : null } +export function getSystemEventByDedupeKey( + db: Database, + dedupeKey: string, + namespace = 'default' +): StoredSystemEvent | null { + const row = db.prepare( + 'SELECT * FROM events WHERE dedupe_key = ? AND namespace = ? LIMIT 1' + ).get(dedupeKey, namespace) as SystemEventRow | undefined + return row ? mapRow(row) : null +} + /** * Read-only extended event query for the Overseer. Additive over * {@link listSystemEvents}; the existing route/promotion paths are untouched. @@ -340,6 +361,34 @@ export function countSystemEvents(db: Database): number { * Idempotent Overseer events DDL — runs on every Store init, NOT gated on SCHEMA_VERSION. * Additive Overseer tables must never own a version step (soup composability). */ +function getEventsColumnNames(db: Database): Set { + const rows = db.prepare('PRAGMA table_info(events)').all() as Array<{ name: string }> + return new Set(rows.map((row) => row.name)) +} + +function ensureEventsNamespaceColumn(db: Database): void { + const columns = getEventsColumnNames(db) + if (columns.size === 0) { + return + } + if (!columns.has('namespace')) { + db.exec(`ALTER TABLE events ADD COLUMN namespace TEXT NOT NULL DEFAULT 'default'`) + db.exec(` + UPDATE events + SET namespace = COALESCE( + (SELECT sessions.namespace FROM sessions WHERE sessions.id = events.related_session_id), + namespace + ) + WHERE related_session_id IS NOT NULL + `) + } + db.exec(` + DROP INDEX IF EXISTS idx_events_dedupe_key; + CREATE UNIQUE INDEX IF NOT EXISTS idx_events_namespace_dedupe_key + ON events(namespace, dedupe_key) WHERE dedupe_key IS NOT NULL; + `) +} + export function ensureOverseerEventsSchema(db: Database): void { db.exec(` CREATE TABLE IF NOT EXISTS events ( @@ -364,11 +413,11 @@ export function ensureOverseerEventsSchema(db: Database): void { provenance TEXT, idempotency_key TEXT, confidence REAL, - severity INTEGER + severity INTEGER, + namespace TEXT NOT NULL DEFAULT 'default' ); CREATE INDEX IF NOT EXISTS idx_events_session_ts ON events(related_session_id, ts DESC); CREATE INDEX IF NOT EXISTS idx_events_type_ts ON events(event_type, ts DESC); - CREATE UNIQUE INDEX IF NOT EXISTS idx_events_dedupe_key ON events(dedupe_key) WHERE dedupe_key IS NOT NULL; CREATE UNIQUE INDEX IF NOT EXISTS idx_events_idempotency_key ON events(idempotency_key) WHERE idempotency_key IS NOT NULL; CREATE TABLE IF NOT EXISTS event_links ( @@ -411,6 +460,8 @@ export function ensureOverseerEventsSchema(db: Database): void { VALUES (new.id, new.summary, COALESCE(new.tags, ''), COALESCE(new.payload_json, '')); END; `) + + ensureEventsNamespaceColumn(db) } /** @deprecated use ensureOverseerEventsSchema */ @@ -466,6 +517,7 @@ export function dropOverseerEventsSchema(db: Database): void { DROP INDEX IF EXISTS idx_event_links_to; DROP INDEX IF EXISTS idx_event_links_from; DROP TABLE IF EXISTS event_links; + DROP INDEX IF EXISTS idx_events_namespace_dedupe_key; DROP INDEX IF EXISTS idx_events_dedupe_key; DROP INDEX IF EXISTS idx_events_type_ts; DROP INDEX IF EXISTS idx_events_session_ts; diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index a3f81b1fb3..84b0372058 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -434,17 +434,27 @@ export class SyncEngine { * the operator inbox. Deduped replays return the prior row without * re-promoting or mutating inbox. */ - insertChannelSystemEvent(input: InsertSystemEventInput): { event: StoredSystemEvent; deduped: boolean } | null { + insertChannelSystemEvent( + namespace: string, + input: InsertSystemEventInput + ): { event: StoredSystemEvent; deduped: boolean } | null { if (input.sourceKind !== 'channel') { throw new Error('insertChannelSystemEvent requires sourceKind channel') } - if (input.idempotencyKey) { - const existing = this.store.events.getByIdempotencyKey(input.idempotencyKey) + const namespacedInput = { ...input, namespace } + if (namespacedInput.idempotencyKey) { + const existing = this.store.events.getByIdempotencyKey(namespacedInput.idempotencyKey) + if (existing) { + return { event: existing, deduped: true } + } + } + if (namespacedInput.dedupeKey) { + const existing = this.store.events.getByDedupeKey(namespacedInput.dedupeKey, namespace) if (existing) { return { event: existing, deduped: true } } } - const event = this.store.events.insert(input) + const event = this.store.events.insert(namespacedInput) if (!event) { return null } diff --git a/hub/src/web/routes/systemEvents.test.ts b/hub/src/web/routes/systemEvents.test.ts index b884da7e4f..2882484b08 100644 --- a/hub/src/web/routes/systemEvents.test.ts +++ b/hub/src/web/routes/systemEvents.test.ts @@ -142,6 +142,70 @@ describe('systemEvents routes', () => { expect(body.events[0]?.summary).toBe('CI failed on upstream PR') }) + it('dedupes on same dedupeKey with different idempotencyKey', async () => { + const store = new Store(':memory:') + const session = store.sessions.getOrCreateSession('dedupe-key-sess', { flavor: 'codex', path: '/tmp' }, null, 'default') + const app = createApp(createEngine(store)) + const sharedDedupe = 'exit-reflection:dedupe-key-sess:986' + + const first = await postEvent(app, validChannelBody({ + relatedSessionId: session.id, + summary: 'Exit reflection skip: timebox: long reason', + dedupeKey: sharedDedupe, + idempotencyKey: 'exit-reflection:dedupe-key-sess:986:hash-a' + })) + expect(first.status).toBe(201) + + const second = await postEvent(app, validChannelBody({ + relatedSessionId: session.id, + summary: 'Exit reflection skip: timebox', + dedupeKey: sharedDedupe, + idempotencyKey: 'exit-reflection:dedupe-key-sess:986:hash-b' + })) + expect(second.status).toBe(200) + const secondJson = await second.json() as { event: { id: number }; deduped: boolean } + expect(secondJson.deduped).toBe(true) + + expect(store.events.count()).toBe(1) + }) + + it('scopes dedupeKey to request namespace', async () => { + const store = new Store(':memory:') + const defaultSession = store.sessions.getOrCreateSession( + 'dedupe-ns-default', + { flavor: 'codex', path: '/tmp' }, + null, + 'default' + ) + const otherSession = store.sessions.getOrCreateSession( + 'dedupe-ns-other', + { flavor: 'codex', path: '/tmp' }, + null, + 'other-ns' + ) + const defaultApp = createApp(createEngine(store), 'default') + const otherApp = createApp(createEngine(store), 'other-ns') + const sharedDedupe = 'contrib:tiann/hapi#999:blocked' + + const first = await postEvent(defaultApp, validChannelBody({ + relatedSessionId: defaultSession.id, + dedupeKey: sharedDedupe, + idempotencyKey: 'contrib:tiann/hapi#999:fp-default' + })) + expect(first.status).toBe(201) + + const second = await postEvent(otherApp, validChannelBody({ + relatedSessionId: otherSession.id, + dedupeKey: sharedDedupe, + idempotencyKey: 'contrib:tiann/hapi#999:fp-other' + })) + expect(second.status).toBe(201) + const secondJson = await second.json() as { deduped: boolean } + expect(secondJson.deduped).toBe(false) + + expect(store.events.count()).toBe(2) + }) + it('dedupes on same idempotencyKey', async () => { const store = new Store(':memory:') const app = createApp(createEngine(store)) diff --git a/hub/src/web/routes/systemEvents.ts b/hub/src/web/routes/systemEvents.ts index 758bf4f970..502135372f 100644 --- a/hub/src/web/routes/systemEvents.ts +++ b/hub/src/web/routes/systemEvents.ts @@ -116,7 +116,7 @@ export function createSystemEventsRoutes(getSyncEngine: () => SyncEngine | null) } } - const result = engine.insertChannelSystemEvent({ + const result = engine.insertChannelSystemEvent(c.get('namespace'), { ts: data.ts ?? Date.now(), sourceKind: 'channel', sourceRef: data.sourceRef,