feat(omni): unified executor layer — merge World B into World A - #1062
Conversation
Implements Group 3 of the unified-executor-layer wish: the omni-bridge must start without PG and gracefully degrade on runtime failures so a dropped PG connection never drops a user reply. - Add `pgAvailable: boolean` on OmniBridge and expose it on BridgeStatus - New private `probePg()` runs at start() after NATS connects; never throws, logs warn on failure, and sets pgAvailable=false for degraded mode - New private `safePgCall<T>(op, fn, fallback, ctx?)` helper — single entry point for every downstream PG call. Try-once semantics, 2s runtime timeout, fallback return on error, flips pgAvailable=false on connection-level errors - Inject hooks: `BridgeConfig.pgProvider` and `BridgeConfig.natsConnectFn` for hermetic unit tests (default to lib/db.getConnection() and nats.connect) - Helper utilities: `withTimeout` and `isPgConnectionError` classify failures by postgres.js code and common message fragments - Tests in src/services/__tests__/omni-bridge.test.ts cover: degraded startup (provider throws, SELECT 1 fails), mid-run connection loss flips the flag, non-connection errors keep the flag true, fast-path short-circuit when PG was never healthy, and the happy path forwards the fn result. Downstream groups (4, 5, 6, 7) will wire their PG writes through safePgCall.
…low-query test
Closes the three gaps flagged in WISH Post-Audit Decision 3 before Wave 2
dispatches. Groups 4/5/6/7 depend on these changes to start.
1. `safePgCall` is now `public` on `OmniBridge` (Decision 2). Downstream
executors hold an `OmniBridge` reference and call the method directly,
e.g. `bridge.safePgCall('op', fn, fallback, { chatId })`. Tests updated
to drop the `(bridge as any)` workaround.
2. `probePg()` now classifies startup errors (Decision 3):
- Connection-level (ECONNREFUSED, connection terminated, …) → degrade
gracefully, log warn, `pgAvailable=false`. Dev without PG still works.
- Anything else (schema mismatch, missing relation, permission denied)
→ fail-fast with an actionable error pointing at the migration
command. Matches the wish's PG Error Handling Strategy table row for
"Migration missing / schema mismatch" — silent corruption is worse
than noisy startup failure.
3. New unit tests in `src/services/__tests__/omni-bridge.test.ts`:
- Fail-fast path: SELECT 1 rejects with "relation 'sessions' does not
exist" → `start()` throws with "PG schema mismatch … migrate" hint.
- Fail-fast variant: provider throws a non-connection error with
code 42501 (permission denied) → `start()` throws with hint.
- Slow-query fallback: fn delays 2.5s (beyond the 2s runtime budget)
→ `safePgCall` returns fallback, `pgAvailable` STAYS true (slow is
not a connection loss), elapsed time confirms the budget held.
Validation:
bun run typecheck → clean
bun run lint → 0 errors
bun test src/services/__tests__/omni-bridge.test.ts → 9 pass / 31 expects
Group 5 — SDK sessions now produce the same sessions + session_content rows that session-capture.ts (filewatch) produces for tmux sessions. New files: - src/lib/audit-events.ts: shared AuditEventType union + recordAuditEvent helper (writes to audit_events via safePgCall, used by Group 5 & 7) - src/services/executors/sdk-session-capture.ts: startSession, recordTurn, updateTurnCount, endSession — all writes through SafePgCallFn - src/services/executors/__tests__/sdk-session-capture.test.ts: 15 tests Integration in claude-sdk.ts _processDelivery(): - deliver.start / deliver.end audit events bracketing each query - User + assistant turns recorded in session_content per delivery - PG session row created lazily on first delivery, ended on shutdown - All writes skip silently when safePgCall is null (degraded mode)
…sdk test mocks - Add _sql parameter to happySafePgCall/degradedSafePgCall to match SafePgCallFn type - Add mockSql tagged-template stub so audit-events callbacks don't crash - Add missing findLatestByMetadata mock to executor-registry mock - Add mock parameter types to createAndLinkExecutorMock for call destructuring - Restore SDK query mock in concurrent delivery afterEach to prevent test pollution
Add bridge restart recovery for in-flight Claude sessions: - Migration 026: partial JSONB index on executors(agent_id, source, chat_id) for fast omni-sourced executor lookups (WHERE ended_at IS NULL) - executor-registry: add findLatestByMetadata, relinkExecutorToAgent, updateClaudeSessionId helpers - claude-sdk spawn(): before creating fresh executor, look up existing live executor via findLatestByMetadata and reuse it (with session ID) - claude-sdk deliver(): detect resume rejection (SDK returns different session ID), persist new session ID, write session.resume_rejected audit - Audit events: session.resumed, session.created_fresh, session.resume_rejected - 11 new tests covering all resume paths in claude-sdk-resume.test.ts
Remove .genie/wishes/ from .gitignore so wish plans and audit artifacts are version-controlled alongside the code they describe.
SafePgCallFn type, safePgCall injection, tmux World A registration — left uncommitted by dead agents.
…p semantics status() now queries the executors table for active omni session count and executor IDs when PG is available, falling back to the local Map size in degraded mode. BridgeStatus gains an executorIds field. The sessions Map is documented as holding process-local runtime handles only (executor instances, idle timers, message buffers) — PG via executor-registry is the source of truth for session identity/state. All existing status() callers updated to await the now-async method. Two new tests cover PG-backed and degraded-mode status paths.
Status now shows: bridge state, NATS connection, pgAvailable, executor type (tmux/sdk), PG-backed active executor count with fallback to in-memory, and executor IDs from PG. Extracted printStatus helper to reduce cognitive complexity.
Add resolveExecutorType() with clear precedence: 1. CLI --executor flag (override) 2. GENIE_EXECUTOR env var (replaces GENIE_EXECUTOR_TYPE) 3. Persisted config (~/.genie/config.json → omni.executor) 4. Default: 'tmux' - New: src/lib/executor-config.ts — single resolver function - Modified: omni-bridge constructor uses resolver instead of inline logic - Modified: genie omni start --executor passes through to resolver - Modified: genie omni status shows resolved executor type - Added executor field to OmniConfigSchema for persistent config - 10 tests covering precedence, invalid values, and fallbacks
…path Extract SafePgCallFn to lib/safe-pg-call.ts (fixes cross-layer import from lib/audit-events → services/executor). Slim executor.ts from 99 to 43 lines by removing verbose JSDoc and section separators. Add TODO markers on OmniSession and IExecutor pointing to World A replacements. OmniSession references now confined to executor.ts + 3 direct consumers (omni-bridge, claude-code executor, claude-sdk executor) — no leakage outside src/services/.
Bridge is a message source, not a state owner. All session state lives in the existing executors/sessions tables.
Add audit-events.ts and executor-config.ts as entry points. Remove stale ignoreBinaries (which) and ignoreDependencies (esbuild).
Group 11 accidentally deleted this new test file (mistaking it for World B). Also sync knip.json after linter adjustments.
…tests claude-sdk-resume.test.ts used mock.module for audit-events.js and sdk-session-capture.js. bun's mock.module is process-global and leaked into sdk-session-capture.test.ts, causing 14 failures in the full suite while passing in isolation. Fix: remove both mock.module calls from the resume test. Instead: - sdk-session-capture runs unmodified (safePgCall is wired with a tracking fake sql that returns row-shaped results) - audit events are tracked via safePgCall's op log (op = 'audit:<type>') instead of a mock.module spy on recordAuditEvent - The rejection audit test extracts attrs from the serialized JSON in the safePgCall SQL values Result: 2127 pass, 0 fail across 111 files (matching dev's 0-fail baseline).
Prevents bun mock.module registrations (6 calls) from leaking into subsequent test files. Same pattern as claude-sdk-resume.test.ts fix.
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements the "Unified Executor Layer" by merging the omni bridge's session management into the core executor registry. It introduces a metadata index for session lookups, lazy resume for Claude sessions, and a runtime switch to choose between tmux and sdk executors. Additionally, it adds a safePgCall utility for graceful degradation when PostgreSQL is offline and implements inline session content capture for the SDK executor. Feedback suggests using undefined instead of empty strings for missing audit event IDs to ensure proper database NULL values and recommends consolidating duplicated ls command definitions to improve maintainability.
| 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) ?? ''; |
There was a problem hiding this comment.
For consistency with how actor is handled (defaulting to null), it would be better for entityId to default to a value that results in NULL in the database, rather than an empty string ''. Using undefined will achieve this with postgres.js and is generally a better representation for a missing value in a database context. This also aligns with the optional nature of executorId in SafePgCallContext.
| const entityId = (attrs.entity_id as string) ?? (attrs.executor_id as string) ?? ''; | |
| const entityId = (attrs.entity_id as string) ?? (attrs.executor_id as string); |
| .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 }) => { |
There was a problem hiding this comment.
The ls command is also defined in src/term-commands/agent/list.ts. Having the command definition in two places can lead to inconsistencies and maintenance issues. It would be better to consolidate the definition in one place, likely src/term-commands/agent/list.ts, and have genie.ts just register it. If this consolidation introduces a circular dependency, use dynamic imports (require() or import()) to break the cycle as per repository guidelines.
References
- Use dynamic imports (
require()orimport()) to break circular dependency cycles that would otherwise occur at module load time.
| .alias('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 }) => { |
There was a problem hiding this comment.
The ls command is also defined in src/genie.ts. This duplication can cause maintenance problems. It's recommended to have a single source of truth for the command definition, probably in this file, and remove the definition from genie.ts. If this consolidation introduces a circular dependency, use dynamic imports (require() or import()) to break the cycle as per repository guidelines.
References
- Use dynamic imports (
require()orimport()) to break circular dependency cycles that would otherwise occur at module load time.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 47f4364c26
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if (claudeSessionId && state.dbSessionId?.startsWith('sdk-')) { | ||
| const newId = await startSession(this.safePgCall, state.executorId ?? sessionKey, claudeSessionId, agentName); | ||
| if (newId) state.dbSessionId = newId; |
There was a problem hiding this comment.
When a fresh SDK chat starts without a Claude session ID, captureUserTurn writes the first user message into a synthetic sdk-* session row; then this block creates a second row keyed by the real Claude session ID and switches state.dbSessionId to it. Because no migration/update happens, the transcript is split across two sessions IDs, total_turns is updated on the new row only, and shutdown later ends only the new row, leaving the synthetic row active/orphaned. This corrupts session history for the most common first-turn path.
Useful? React with 👍 / 👎.
| // State: working (query in flight) — before the blocking operation starts. | ||
| if (state.executorId) await this.updateState(state.executorId, 'working', session.chatId); |
There was a problem hiding this comment.
This method marks the executor as working before running the query, but idle is only written on the success path; any exception from runQuery, stream consumption, capture writes, or publish exits early and never performs the working → idle transition. Since deliver() intentionally swallows queue errors, these failures do not propagate to a caller that could repair state, so executors can remain stuck as working indefinitely after transient delivery errors.
Useful? React with 👍 / 👎.
namastex888
changed the title
Audit all 80+ wishes against dev codebase. 17 had stale statuses: SHIPPED (14): - unified-omni-bridge (PRs #1063, #1065) - fix-omni-bridge-hardening (PR #1065) - unified-executor-layer (PR #1062) - auto-orchestrate, fix-depends-parser, parallel-execution - task-projects, test-pg-ram-isolation, task-auto-close-on-merge - worktree-out-of-repo, docs-overhaul, genie-hacks-community-docs - multi-agent-session-isolation, session-auto-create OBSOLETE (3): - genie-omni-marriage (superseded by smaller wishes) - fix-session-uuid-resume (replaced by --continue by name) - qa-dev-to-main (time-bound QA from March 20)
1 participant
Summary
Merges the genie CLI's two parallel session/executor subsystems into a single unified layer:
executor-registry.ts, PG-backedsessions/executors/session_content/audit_eventstables,session-capture.tsfilewatchservices/executor.tsIExecutor interface, in-memory session MapAfter this PR, World B executors (claude-sdk, claude-code) register in World A on spawn, update state on deliver, and terminate on shutdown — making them visible to
genie ls,genie sessions, audit trails, and the session observatory.Supersedes PR #1042 with a proper 12-group wish-driven implementation.
Key changes
pgAvailable=falseon connection errors)claude-sdk.tsandclaude-code.tscallfindOrCreateAgent+createAndLinkExecutoron spawn,updateExecutorStateon deliver,terminateExecutoron shutdownAuditEventTypeenum with dotted naming (executor.spawn,deliver.start,session.resumed, etc.) insrc/lib/audit-events.tssdk-session-capture.tswrites SDK conversation turns tosession_contentvia safePgCallfindLatestByMetadata({agentId, source, chatId})queries executors table with new partial JSONB index for fast omni-context lookupsexecutor-config.tswith precedence: CLI override > env > persisted config > default 'tmux'genie omni statusshows pgAvailable + executor count;genie sessions/ls/execgain--sourcefilter026_executors_omni_metadata_index.sqladds partial index on(agent_id, metadata->>'source', metadata->>'chat_id') WHERE ended_at IS NULLservices/executor.tsdocumented as compatibility shim; elimination path chartedStats
Test plan
tsc --noEmit— cleanbiome check .— clean (pre-existing warnings only)bunx knip— cleanbun test— 2127/0 (worktree), 2040/0 (main repo)genie omni statusshows pgAvailable + active executor countgenie ls --source omnifilters to omni-bridge spawned executorsgenie sessions listafter omni-bridge spawn