fix(hub): dedupe channel SystemEvents on dedupe_key before INSERT

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)
     }

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')
+    })
+})

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<string> {
+    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;

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
         }

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))

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,