fix(OMN-12006): move event bus config to contract overlay

.gitignore

@@ -1,7 +1,9 @@
 node_modules/
+node_modules
 dist/
 .vite/
 *.local
+contract.local.yaml
 .env.local*
 fixtures/
 # Storybook fixture builders are part of the source tree (OMN-103).

contract.local.yaml.example

@@ -0,0 +1,7 @@
+# Copy to contract.local.yaml to select a concrete runtime lane.
+# contract.local.yaml is ignored and may contain LAN or Tailscale endpoints.
+
+event_bus:
+  # Stability-test Redpanda from Jonah's Mac.
+  bootstrap_servers: "192.168.86.201:39092" # onex-allow-internal-ip
+  client_id: "omnidash-server"

contract.yaml

@@ -29,6 +29,13 @@ data_source:
   # Tilde is expanded to $HOME by the server at runtime.
   sqlite_db_path: "~/.omninode/delegation/delegation.sqlite"
 
+event_bus:
+  # Contract-owned Kafka/Redpanda bootstrap configuration for the Express
+  # dispatch bridge. Keep lane-specific endpoints in contract.local.yaml or
+  # deployment overlays; the base contract intentionally has no network target.
+  bootstrap_servers: ""
+  client_id: "omnidash-server"
+
 command_topics:
   # OMN-12070: Delegation command dispatched via the Express trigger bridge
   # to the stability-test runtime effects endpoint (OMNIDASH_RUNTIME_EFFECTS_URL).

scripts/check-no-env-contamination.sh

@@ -76,6 +76,8 @@ ALLOWLIST_FILES=(
   # default for the data-source bridge URL. These are defaults, not hardcoded
   # environment-specific literals.
   'contract.yaml'
+  'contract.local.yaml'
+  'contract.local.yaml.example'
   'server/data-source-contract.ts'
   'scripts/generate-data-source-config.ts'
   'server/__tests__/data-source-contract.test.ts'

scripts/generate-data-source-config.ts

@@ -7,12 +7,13 @@
  * remain as optional runtime overrides.
  */
 
-import { readFileSync, mkdirSync, writeFileSync } from 'node:fs';
+import { existsSync, readFileSync, mkdirSync, writeFileSync } from 'node:fs';
 import { resolve, dirname } from 'node:path';
 import { fileURLToPath } from 'node:url';
 
 const __dirname = dirname(fileURLToPath(import.meta.url));
 const contractPath = resolve(__dirname, '..', 'contract.yaml');
+const contractOverlayPath = resolve(__dirname, '..', 'contract.local.yaml');
 const outPath = resolve(__dirname, '..', 'src', 'config', 'generated', 'data-source-defaults.ts');
 
 type DataSourceMode = 'sqlite' | 'postgres' | 'file' | 'http';
@@ -24,13 +25,17 @@ interface Defaults {
   sqliteDbPath: string;
 }
 
-function parseContract(raw: string): Defaults {
-  const defaults: Defaults = {
+function defaultDataSource(): Defaults {
+  return {
     mode: 'sqlite',
     url: 'http://localhost:3002',
     wsUrl: 'ws://localhost:3002/ws',
     sqliteDbPath: '~/.omninode/delegation/delegation.sqlite',
   };
+}
+
+function parseContract(raw: string, initial: Defaults = defaultDataSource()): Defaults {
+  const defaults: Defaults = { ...initial };
 
   let inDataSource = false;
   for (const rawLine of raw.split('\n')) {
@@ -58,8 +63,10 @@ function parseContract(raw: string): Defaults {
   return defaults;
 }
 
-const raw = readFileSync(contractPath, 'utf8');
-const d = parseContract(raw);
+let d = parseContract(readFileSync(contractPath, 'utf8'));
+if (existsSync(contractOverlayPath)) {
+  d = parseContract(readFileSync(contractOverlayPath, 'utf8'), d);
+}
 
 const generated = `// GENERATED — do not edit. Source: contract.yaml
 // Regenerate with: npm run generate:config

server/__tests__/data-source-contract.test.ts

@@ -1,5 +1,8 @@
 import { describe, it, expect, beforeEach, afterEach } from 'vitest';
-import { loadDataSourceConfig } from '../data-source-contract.js';
+import { mkdtempSync, rmSync, writeFileSync, readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { tmpdir } from 'node:os';
+import { loadDataSourceConfig, loadRuntimeContract } from '../data-source-contract.js';
 
 describe('loadDataSourceConfig', () => {
   const savedEnv: Record<string, string | undefined> = {};
@@ -51,4 +54,40 @@ describe('loadDataSourceConfig', () => {
     expect(cfg.sqliteDbPath).not.toMatch(/^~/);
     expect(cfg.sqliteDbPath).toMatch(/\.omninode/);
   });
+
+  it('deep-merges contract.local.yaml over contract.yaml without resetting unrelated defaults', () => {
+    const dir = mkdtempSync(join(tmpdir(), 'omnidash-contract-'));
+    try {
+      const basePath = join(dir, 'contract.yaml');
+      const overlayPath = join(dir, 'contract.local.yaml');
+      writeFileSync(basePath, `
+data_source:
+  default: postgres
+  url: "http://base:3002"
+  ws_url: "ws://base:3002/ws"
+  sqlite_db_path: "~/.omninode/delegation/delegation.sqlite"
+event_bus:
+  bootstrap_servers: ""
+  client_id: "omnidash-server"
+`);
+      writeFileSync(overlayPath, `
+event_bus:
+  bootstrap_servers: "192.168.86.201:39092"
+`);
+
+      const cfg = loadRuntimeContract(basePath, overlayPath);
+
+      expect(cfg.data_source.default).toBe('postgres');
+      expect(cfg.data_source.url).toBe('http://base:3002');
+      expect(cfg.event_bus.bootstrap_servers).toBe('192.168.86.201:39092');
+      expect(cfg.event_bus.client_id).toBe('omnidash-server');
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+
+  it('does not use legacy KAFKA_BROKERS in the producer config path', () => {
+    const source = readFileSync(join(process.cwd(), 'server', 'kafka-producer.ts'), 'utf8');
+    expect(source).not.toContain('KAFKA_BROKERS');
+  });
 });

server/data-source-contract.ts

@@ -1,71 +1,118 @@
-import { readFileSync } from 'node:fs';
+import { existsSync, readFileSync } from 'node:fs';
 import { resolve, join } from 'node:path';
 import { homedir } from 'node:os';
 import { fileURLToPath } from 'node:url';
 
 const __dirname = resolve(fileURLToPath(import.meta.url), '..');
 const CONTRACT_PATH = resolve(__dirname, '..', 'contract.yaml');
+const CONTRACT_OVERLAY_PATH = resolve(__dirname, '..', 'contract.local.yaml');
 
 export type DataSourceMode = 'sqlite' | 'postgres' | 'file' | 'http';
 
-interface DataSourceContract {
+interface RuntimeContract {
   data_source: {
     default: DataSourceMode;
     url: string;
     ws_url: string;
     sqlite_db_path: string;
   };
+  event_bus: {
+    bootstrap_servers: string;
+    client_id: string;
+  };
 }
 
-function parseYamlDataSource(raw: string): DataSourceContract {
-  // Minimal YAML parser for the flat contract.yaml structure.
-  // js-yaml is a transitive dep but not a declared direct dep — use a
-  // simple line-by-line parse to avoid coupling to an undeclared package.
-  const result: DataSourceContract = {
+type RuntimeContractPatch = {
+  data_source?: Partial<RuntimeContract['data_source']>;
+  event_bus?: Partial<RuntimeContract['event_bus']>;
+};
+
+function defaultContract(): RuntimeContract {
+  return {
     data_source: {
       default: 'sqlite',
       url: 'http://localhost:3002',
       ws_url: 'ws://localhost:3002/ws',
       sqlite_db_path: '~/.omninode/delegation/delegation.sqlite',
     },
+    event_bus: {
+      bootstrap_servers: '',
+      client_id: 'omnidash-server',
+    },
   };
+}
+
+function parseYamlRuntimeContract(raw: string): RuntimeContractPatch {
+  // Minimal YAML parser for the flat contract.yaml runtime config sections.
+  // js-yaml is a transitive dep but not a declared direct dep — use a
+  // simple line-by-line parse to avoid coupling to an undeclared package.
+  const result: RuntimeContractPatch = {};
+  let section: keyof RuntimeContract | null = null;
 
-  let inDataSource = false;
   for (const rawLine of raw.split('\n')) {
     const line = rawLine.replace(/#.*$/, '').trimEnd();
     if (!line.trim()) continue;
 
-    if (line === 'data_source:') {
-      inDataSource = true;
+    const sectionMatch = line.match(/^(\w+):\s*$/);
+    if (sectionMatch) {
+      const name = sectionMatch[1] as keyof RuntimeContract;
+      section = name === 'data_source' || name === 'event_bus' ? name : null;
       continue;
     }
-    if (inDataSource && /^\S/.test(line) && !line.startsWith(' ') && !line.startsWith('\t')) {
-      inDataSource = false;
-    }
-    if (!inDataSource) continue;
+    if (!section) continue;
 
     const m = line.match(/^\s+(\w+):\s+"?([^"]*)"?\s*$/);
     if (!m) continue;
     const [, key, value] = m;
-    if (key === 'default') result.data_source.default = value as DataSourceMode;
-    else if (key === 'url') result.data_source.url = value;
-    else if (key === 'ws_url') result.data_source.ws_url = value;
-    else if (key === 'sqlite_db_path') result.data_source.sqlite_db_path = value;
+    if (section === 'data_source') {
+      result.data_source ??= {};
+      if (key === 'default') result.data_source.default = value as DataSourceMode;
+      else if (key === 'url') result.data_source.url = value;
+      else if (key === 'ws_url') result.data_source.ws_url = value;
+      else if (key === 'sqlite_db_path') result.data_source.sqlite_db_path = value;
+    } else if (section === 'event_bus') {
+      result.event_bus ??= {};
+      if (key === 'bootstrap_servers') result.event_bus.bootstrap_servers = value;
+      else if (key === 'client_id') result.event_bus.client_id = value;
+    }
   }
 
   return result;
 }
 
+function mergeContract(base: RuntimeContract, overlay: RuntimeContractPatch): RuntimeContract {
+  return {
+    data_source: {
+      ...base.data_source,
+      ...overlay.data_source,
+    },
+    event_bus: {
+      ...base.event_bus,
+      ...overlay.event_bus,
+    },
+  };
+}
+
 function expandTilde(p: string): string {
   return p.startsWith('~') ? join(homedir(), p.slice(1)) : p;
 }
 
-let _contract: DataSourceContract | null = null;
+let _contract: RuntimeContract | null = null;
+
+export function loadRuntimeContract(
+  contractPath = CONTRACT_PATH,
+  overlayPath = CONTRACT_OVERLAY_PATH,
+): RuntimeContract {
+  let contract = mergeContract(defaultContract(), parseYamlRuntimeContract(readFileSync(contractPath, 'utf8')));
+  if (existsSync(overlayPath)) {
+    contract = mergeContract(contract, parseYamlRuntimeContract(readFileSync(overlayPath, 'utf8')));
+  }
+  return contract;
+}
 
-function loadContract(): DataSourceContract {
+function loadContract(): RuntimeContract {
   if (_contract) return _contract;
-  const raw = readFileSync(CONTRACT_PATH, 'utf8');
-  _contract = parseYamlDataSource(raw);
+  _contract = loadRuntimeContract();
   return _contract;
 }
 
@@ -94,3 +141,22 @@ export function loadDataSourceConfig(): DataSourceConfig {
     sqliteDbPath: expandTilde(rawDbPath),
   };
 }
+
+export interface EventBusConfig {
+  /** Contract-resolved Kafka/Redpanda broker list for dispatch publishing. */
+  bootstrapServers: string[];
+  /** Kafka client id for the omnidash bridge producer. */
+  clientId: string;
+}
+
+export function loadEventBusConfig(): EventBusConfig {
+  const contract = loadContract();
+  const bootstrapServers = contract.event_bus.bootstrap_servers
+    .split(',')
+    .map((value) => value.trim())
+    .filter(Boolean);
+  return {
+    bootstrapServers,
+    clientId: contract.event_bus.client_id,
+  };
+}

server/kafka-producer.ts

@@ -1,17 +1,19 @@
 import { Kafka, Producer, logLevel } from 'kafkajs';
-
-const KAFKA_BROKERS = (process.env.KAFKA_BROKERS ?? 'localhost:19092').split(',');
-
-const kafka = new Kafka({
-  clientId: 'omnidash-server',
-  brokers: KAFKA_BROKERS,
-  logLevel: logLevel.WARN,
-});
+import { loadEventBusConfig } from './data-source-contract.js';
 
 let producer: Producer | null = null;
 let connected = false;
 
 export async function connectProducer(): Promise<void> {
+  const config = loadEventBusConfig();
+  if (config.bootstrapServers.length === 0) {
+    throw new Error('event_bus.bootstrap_servers missing; configure contract.local.yaml or a deployment overlay');
+  }
+  const kafka = new Kafka({
+    clientId: config.clientId,
+    brokers: config.bootstrapServers,
+    logLevel: logLevel.WARN,
+  });
   producer = kafka.producer();
   await producer.connect();
   connected = true;

src/components/dashboard/cost-savings-overview/CostSavingsOverviewWidget.test.tsx

@@ -116,7 +116,7 @@ describe('CostSavingsOverviewWidget', () => {
     expect(await screen.findByText('Delegated Tokens')).toBeInTheDocument();
     expect(screen.getByText('1,202 to compliance')).toBeInTheDocument();
     expect(screen.getAllByText('Qwen3-Coder-30B-A3B-Instruct-AWQ-4bit').length).toBeGreaterThan(0);
-    expect(screen.getByText('+$0.0153')).toBeInTheDocument();
+    expect(screen.getByText('+$0.02')).toBeInTheDocument();
     expect(screen.getByText('100.0%')).toBeInTheDocument();
     expect(screen.getByText('Recent delegation runs')).toBeInTheDocument();
     expect(screen.getByText('5e12c850')).toBeInTheDocument();

src/components/dashboard/cost-savings-overview/CostSavingsOverviewWidget.tsx

@@ -83,7 +83,7 @@ const MODE_COLOR: Record<ExecutionMode, string> = {
 };
 
 function fmtUsd(v: number): string {
-  return v === 0 ? '$0.00' : `$${v.toFixed(Math.abs(v) < 1 ? 4 : 2)}`;
+  return `$${v.toFixed(2)}`;
 }
 
 function fmtPct(v: number): string {
@@ -144,7 +144,7 @@ function KPIBar({ overview }: { overview: CostSavingsOverviewProjection }) {
         label="Cloud Avoided"
         value={overview.total_savings_usd}
         prefix="$"
-        decimals={Math.abs(overview.total_savings_usd) < 1 ? 4 : 2}
+        decimals={2}
         tone={savingsTone}
       />
       <KPI

src/components/dashboard/delegation/DelegationCostComparisonWidget.tsx

@@ -19,7 +19,7 @@ export interface DelegationCostComparisonConfig {
 
 function fmtUsd(v: number | undefined): string {
   if (v == null || !Number.isFinite(v)) return '—';
-  return v === 0 ? '$0.00' : `$${v.toFixed(v < 0.01 ? 4 : 2)}`;
+  return `$${v.toFixed(2)}`;
 }
 
 function fmtTokens(v: number | undefined): string {
@@ -139,9 +139,9 @@ function SummaryBar({ projection }: { projection: DelegationSavingsProjection })
 
   return (
     <div style={{ display: 'flex', gap: 10, flexWrap: 'wrap', marginBottom: 12 }}>
-      <KPI label="Cumulative savings" value={projection.cumulative_savings_usd} prefix="$" decimals={4} tone="good" />
-      <KPI label="Local cost" value={projection.cumulative_local_cost_usd} prefix="$" decimals={4} />
-      <KPI label="Cloud baseline" value={projection.cumulative_cloud_cost_usd} prefix="$" decimals={4} />
+      <KPI label="Cumulative savings" value={projection.cumulative_savings_usd} prefix="$" decimals={2} tone="good" />
+      <KPI label="Local cost" value={projection.cumulative_local_cost_usd} prefix="$" decimals={2} />
+      <KPI label="Cloud baseline" value={projection.cumulative_cloud_cost_usd} prefix="$" decimals={2} />
       <KPI label="Sessions" value={projection.session_count} />
       <KPI label="Total tokens" value={totalTokens} />
       <KPI label="Measured" value={measuredCount} caption={`${estimatedCount} estimated`} tone={measuredCount > 0 ? 'good' : 'warn'} />

src/components/dashboard/delegation/DelegationSavingsWidget.test.tsx

@@ -131,7 +131,7 @@ describe('DelegationSavingsWidget', () => {
     expect(await screen.findByText('Delegated tokens')).toBeInTheDocument();
     expect(screen.getByText('1,202 to compliance')).toBeInTheDocument();
     expect(screen.getByTitle('input 144, output 593, compliance 737')).toHaveTextContent('737');
-    expect(screen.getByText('+$0.0093')).toBeInTheDocument();
+    expect(screen.getAllByText('+$0.01').length).toBeGreaterThan(0);
     expect(screen.getByText(/document .*sess-liv/i)).toBeInTheDocument();
   });