feat(omni): unified executor layer — merge World B into World A

.genie/wishes/genie-omni-marriage/WISH.md

@@ -297,6 +297,10 @@ pm2 start ecosystem.config.cjs && pm2 ls | grep genie
 
 ---
 
+## Reconciliation Notes
+
+> **Omni-bridge is a message source, not a state owner.** The omni-bridge relay delivers inbound messages and publishes replies via NATS, but session lifecycle (spawn, resume, freeze, registry) is owned entirely by genie's executor layer. PR #1042 attempted to add a PG-backed session registry inside the bridge — this created a third source of truth alongside the executor table and the worker registry. The unified-executor-layer branch resolves this by keeping session persistence in the executors table and treating the bridge as a stateless message transport. Bridge restarts no longer require session recovery because the bridge holds no session state.
+
 ## Files to Create/Modify
 
 ```

.gitignore

@@ -58,7 +58,6 @@ yarn-error.log*
 .genie/agents.json
 .genie/teams/
 .genie/brainstorms/
-.genie/wishes/
 
 # Team runtime state
 .genie/mailbox/

knip.json

@@ -5,6 +5,7 @@
     "src/tui/render.tsx",
     "src/tui/tmux.ts",
     "src/lib/executor-types.ts",
+    "src/lib/safe-pg-call.ts",
     "src/lib/executor-registry.ts",
     "src/lib/assignment-registry.ts",
     "src/lib/providers/*.ts",
@@ -20,14 +21,15 @@
     "src/term-commands/exec/index.ts",
     "src/term-commands/brief.ts",
     "src/services/omni-reply.ts",
-    "src/services/omni-bridge.ts"
+    "src/services/omni-bridge.ts",
+    "src/lib/audit-events.ts",
+    "src/lib/executor-config.ts"
   ],
   "project": ["src/**/*.ts", "src/**/*.tsx"],
-  "ignoreBinaries": ["tmux", "which"],
+  "ignoreBinaries": ["which"],
   "ignoreExportsUsedInFile": false,
   "ignoreDependencies": [
     "@automagik/genie-brain",
-    "esbuild",
     "@tauri-apps/cli",
     "@tauri-apps/api",
     "react-dom",

src/db/migrations/026_executors_omni_metadata_index.sql

@@ -0,0 +1,11 @@
+-- Partial index for omni-sourced executor lookups.
+-- Covers: findLatestByMetadata in executor-registry (lazy resume),
+--         bridge's find-or-create on inbound NATS message,
+--         genie ls --source omni / genie sessions list --source omni.
+CREATE INDEX IF NOT EXISTS executors_omni_lookup
+  ON executors (
+    agent_id,
+    (metadata->>'source'),
+    (metadata->>'chat_id')
+  )
+  WHERE ended_at IS NULL;

src/genie.ts

@@ -487,7 +487,8 @@ program
   .command('ls')
   .description('List registered agents with runtime status')
   .option('--json', 'Output as JSON')
-  .action(async (options: { json?: boolean }) => {
+  .option('--source <name>', 'Filter by executor metadata source (e.g. omni)')
+  .action(async (options: { json?: boolean; source?: string }) => {
     try {
       await handleLsCommand(options);
     } catch (error) {

src/lib/__tests__/executor-config.test.ts

@@ -0,0 +1,67 @@
+import { afterEach, beforeEach, describe, expect, test } from 'bun:test';
+import { resolveExecutorType } from '../executor-config.js';
+
+describe('resolveExecutorType', () => {
+  let originalEnv: string | undefined;
+
+  beforeEach(() => {
+    originalEnv = process.env.GENIE_EXECUTOR;
+    process.env.GENIE_EXECUTOR = undefined as unknown as string;
+  });
+
+  afterEach(() => {
+    if (originalEnv !== undefined) {
+      process.env.GENIE_EXECUTOR = originalEnv;
+    } else {
+      process.env.GENIE_EXECUTOR = undefined as unknown as string;
+    }
+  });
+
+  test('defaults to tmux when nothing is set', () => {
+    expect(resolveExecutorType()).toBe('tmux');
+  });
+
+  test('override argument wins over everything', () => {
+    process.env.GENIE_EXECUTOR = 'tmux';
+    expect(resolveExecutorType('sdk')).toBe('sdk');
+  });
+
+  test('env var is used when no override', () => {
+    process.env.GENIE_EXECUTOR = 'sdk';
+    expect(resolveExecutorType()).toBe('sdk');
+  });
+
+  test('env var tmux is respected', () => {
+    process.env.GENIE_EXECUTOR = 'tmux';
+    expect(resolveExecutorType()).toBe('tmux');
+  });
+
+  test('override beats env var', () => {
+    process.env.GENIE_EXECUTOR = 'sdk';
+    expect(resolveExecutorType('tmux')).toBe('tmux');
+  });
+
+  test('invalid override falls through to env', () => {
+    process.env.GENIE_EXECUTOR = 'sdk';
+    expect(resolveExecutorType('bogus')).toBe('sdk');
+  });
+
+  test('invalid override and no env falls back to tmux', () => {
+    expect(resolveExecutorType('invalid')).toBe('tmux');
+  });
+
+  test('invalid env falls back to tmux', () => {
+    process.env.GENIE_EXECUTOR = 'docker';
+    expect(resolveExecutorType()).toBe('tmux');
+  });
+
+  test('empty string override falls through', () => {
+    process.env.GENIE_EXECUTOR = 'sdk';
+    expect(resolveExecutorType('')).toBe('sdk');
+  });
+
+  test('undefined override falls through', () => {
+    process.env.GENIE_EXECUTOR = 'sdk';
+    expect(resolveExecutorType(undefined)).toBe('sdk');
+  });
+});

src/lib/audit-events.ts

@@ -0,0 +1,64 @@
+/**
+ * Shared audit event types and recording helper.
+ *
+ * Used by the SDK executor (Group 5) and resume logic (Group 7).
+ * Writes to the `audit_events` table via safePgCall so degraded-mode
+ * (no PG) silently skips — no throw, no data loss risk.
+ */
+
+import type { SafePgCallFn } from './safe-pg-call.js';
+
+// ============================================================================
+// Audit event type union
+// ============================================================================
+
+export type AuditEventType =
+  // Lifecycle
+  | 'executor.spawn'
+  | 'executor.shutdown'
+  | 'executor.state_transition'
+  // Delivery
+  | 'deliver.start'
+  | 'deliver.end'
+  | 'deliver.error'
+  | 'deliver.tool_use'
+  // Resume (Group 7)
+  | 'session.resumed'
+  | 'session.resume_rejected'
+  | 'session.created_fresh';
+
+// ============================================================================
+// Record helper
+// ============================================================================
+
+/**
+ * Write an audit event row via safePgCall.
+ *
+ * Maps to the real audit_events schema:
+ *   entity_type, entity_id, event_type, actor, details, created_at
+ *
+ * `entityType` defaults to 'executor' when not provided in attrs.
+ * `entityId` defaults to attrs.executor_id.
+ * `actor` defaults to attrs.agent_id.
+ */
+export async function recordAuditEvent(
+  safePgCall: SafePgCallFn,
+  type: AuditEventType,
+  attrs: Record<string, unknown>,
+): Promise<void> {
+  const entityType = (attrs.entity_type as string) ?? 'executor';
+  const entityId = (attrs.entity_id as string) ?? (attrs.executor_id as string) ?? '';
+  const actor = (attrs.actor as string) ?? (attrs.agent_id as string) ?? null;
+
+  // Strip meta keys from details — they're top-level columns already.
+  const { entity_type: _et, entity_id: _eid, actor: _a, ...details } = attrs;
+
+  await safePgCall(
+    `audit:${type}`,
+    (sql) =>
+      sql`INSERT INTO audit_events (entity_type, entity_id, event_type, actor, details)
+          VALUES (${entityType}, ${entityId}, ${type}, ${actor}, ${JSON.stringify(details)})`,
+    undefined,
+    { executorId: entityId, chatId: (attrs.chat_id as string) ?? '' },
+  );
+}

src/lib/executor-config.ts

@@ -0,0 +1,40 @@
+/**
+ * Executor type resolver.
+ *
+ * Resolution order:
+ *   1. Explicit override argument (CLI --executor flag)
+ *   2. GENIE_EXECUTOR env var
+ *   3. Persisted config (~/.genie/config.json → omni.executor)
+ *   4. Default: 'tmux'
+ */
+
+import { loadGenieConfigSync } from './genie-config.js';
+
+export type ExecutorType = 'tmux' | 'sdk';
+
+const VALID: ReadonlySet<string> = new Set(['tmux', 'sdk']);
+
+function isValid(value: unknown): value is ExecutorType {
+  return typeof value === 'string' && VALID.has(value);
+}
+
+export function resolveExecutorType(override?: string): ExecutorType {
+  // 1. Explicit override
+  if (isValid(override)) return override;
+
+  // 2. Env var
+  const env = process.env.GENIE_EXECUTOR;
+  if (isValid(env)) return env;
+
+  // 3. Persisted config
+  try {
+    const cfg = loadGenieConfigSync();
+    const persisted = cfg.omni?.executor;
+    if (isValid(persisted)) return persisted;
+  } catch {
+    // Config unreadable — fall through
+  }
+
+  // 4. Default
+  return 'tmux';
+}

src/lib/executor-registry.test.ts

@@ -236,6 +236,41 @@ describe.skipIf(!DB_AVAILABLE)('executor-registry', () => {
       expect(list[0].id).toBe(e2.id); // Newest first
       expect(list[1].id).toBe(e1.id);
     });
+
+    test('filters by metadata source', async () => {
+      const a1 = await seedAgent('omni-agent', 'team1');
+      const a2 = await seedAgent('cli-agent', 'team1');
+      await createExecutor(a1, 'claude', 'api', { metadata: { source: 'omni', chat_id: 'c1' } });
+      await createExecutor(a2, 'claude', 'tmux');
+      await createExecutor(a2, 'claude', 'tmux', { metadata: { source: 'cli' } });
+
+      const omniOnly = await listExecutors(undefined, 'omni');
+      expect(omniOnly.length).toBe(1);
+      expect(omniOnly[0].agentId).toBe(a1);
+      expect(omniOnly[0].metadata).toEqual({ source: 'omni', chat_id: 'c1' });
+
+      const cliOnly = await listExecutors(undefined, 'cli');
+      expect(cliOnly.length).toBe(1);
+      expect(cliOnly[0].agentId).toBe(a2);
+
+      // No source returns all
+      const all = await listExecutors();
+      expect(all.length).toBe(3);
+    });
+
+    test('filters by both agent ID and source', async () => {
+      const a1 = await seedAgent('multi-agent', 'team1');
+      await createExecutor(a1, 'claude', 'api', { metadata: { source: 'omni', chat_id: 'c1' } });
+      await createExecutor(a1, 'claude', 'tmux');
+
+      const filtered = await listExecutors(a1, 'omni');
+      expect(filtered.length).toBe(1);
+      expect(filtered[0].metadata).toEqual({ source: 'omni', chat_id: 'c1' });
+
+      // Agent filter alone returns both
+      const allForAgent = await listExecutors(a1);
+      expect(allForAgent.length).toBe(2);
+    });
   });
 
   // ==========================================================================

src/lib/executor-registry.ts

@@ -180,12 +180,16 @@ export async function terminateActiveExecutor(agentId: string): Promise<void> {
   await sql`UPDATE agents SET current_executor_id = NULL WHERE id = ${agentId} AND current_executor_id = ${executorId}`;
 }
 
-/** List executors, optionally filtered by agent ID. */
-export async function listExecutors(agentId?: string): Promise<Executor[]> {
+/** List executors, optionally filtered by agent ID and/or metadata source. */
+export async function listExecutors(agentId?: string, source?: string): Promise<Executor[]> {
   const sql = await getConnection();
-  const rows = agentId
-    ? await sql<ExecutorRow[]>`SELECT * FROM executors WHERE agent_id = ${agentId} ORDER BY started_at DESC`
-    : await sql<ExecutorRow[]>`SELECT * FROM executors ORDER BY started_at DESC`;
+  const rows = await sql<ExecutorRow[]>`
+    SELECT * FROM executors
+    WHERE true
+    ${agentId ? sql`AND agent_id = ${agentId}` : sql``}
+    ${source ? sql`AND metadata->>'source' = ${source}` : sql``}
+    ORDER BY started_at DESC
+  `;
   return rows.map(rowToExecutor);
 }
 
@@ -205,3 +209,39 @@ export async function findExecutorBySession(claudeSessionId: string): Promise<Ex
   `;
   return rows.length > 0 ? rowToExecutor(rows[0]) : null;
 }
+
+/**
+ * Find the latest live executor matching omni metadata.
+ * Used for lazy resume: on bridge restart, look up an existing executor
+ * for this agent + chat combination so we can reuse its Claude session.
+ * Uses the `executors_omni_lookup` partial index (migration 026).
+ */
+export async function findLatestByMetadata(filter: {
+  agentId: string;
+  source: string;
+  chatId: string;
+}): Promise<Executor | null> {
+  const sql = await getConnection();
+  const rows = await sql<ExecutorRow[]>`
+    SELECT * FROM executors
+    WHERE agent_id = ${filter.agentId}
+      AND metadata->>'source' = ${filter.source}
+      AND metadata->>'chat_id' = ${filter.chatId}
+      AND ended_at IS NULL
+    ORDER BY started_at DESC
+    LIMIT 1
+  `;
+  return rows.length > 0 ? rowToExecutor(rows[0]) : null;
+}
+
+/** Relink an existing executor to an agent (set current_executor_id FK). */
+export async function relinkExecutorToAgent(executorId: string, agentId: string): Promise<void> {
+  const sql = await getConnection();
+  await sql`UPDATE agents SET current_executor_id = ${executorId} WHERE id = ${agentId}`;
+}
+
+/** Update the Claude session ID on an executor row. */
+export async function updateClaudeSessionId(executorId: string, sessionId: string): Promise<void> {
+  const sql = await getConnection();
+  await sql`UPDATE executors SET claude_session_id = ${sessionId} WHERE id = ${executorId}`;
+}

src/lib/safe-pg-call.ts

@@ -0,0 +1,23 @@
+/**
+ * SafePgCallFn — Structural type for the bridge's PG-guarded call pattern.
+ *
+ * Extracted from services/executor.ts so that lib/ modules (audit-events,
+ * sdk-session-capture) can reference it without a cross-layer import.
+ */
+
+import type { Sql } from './db.js';
+
+/**
+ * Bound `safePgCall` injected by the bridge into each executor after construction.
+ *
+ * Mirror of `OmniBridge#safePgCall` — declared here as a structural type so
+ * executors can call it without importing `OmniBridge` (avoids a circular
+ * dependency between the bridge and its own executors). Tests pass a plain
+ * function; production wires `bridge.safePgCall.bind(bridge)`.
+ */
+export type SafePgCallFn = <T>(
+  op: string,
+  fn: (sql: Sql) => Promise<T>,
+  fallback: T,
+  ctx?: { executorId?: string; chatId?: string },
+) => Promise<T>;