Repository navigation
feat(OMN-1992): Activate PluginIntelligence in kernel bootstrap via RegistryDomainPlugin - #262
Merged
jonahgabriel merged 9 commits intoFeb 8, 2026
Conversation
…egistryDomainPlugin Replace ~150 lines of inline PostgreSQL pool creation, handler wiring, dispatcher setup, and consumer startup in service_kernel.py with a plugin registry loop that calls the standard ProtocolDomainPlugin lifecycle: should_activate -> initialize -> wire_handlers -> wire_dispatchers -> start_consumers -> shutdown (LIFO). - Register PluginRegistration and PluginIntelligence with RegistryDomainPlugin - Move all registration infrastructure init into PluginRegistration lifecycle methods - Add node_identity and kafka_bootstrap_servers to ModelDomainPluginConfig - Graceful degradation: kernel boots without omniintelligence if not installed
📝 WalkthroughWalkthroughAdds plugin-driven bootstrap with a domain-plugin lifecycle, optional Kafka-backed SnapshotPublisher support propagated through registration wiring, and new config fields ( Changes
Sequence Diagram(s)sequenceDiagram
participant SK as ServiceKernel
participant PR as PluginRegistry
participant DP as DomainPlugin (PluginRegistration)
participant SP as SnapshotPublisher
participant KS as Kafka
SK->>PR: build plugin_config (incl. node_identity, kafka_bootstrap_servers)
SK->>DP: should_activate(plugin_config)
DP-->>SK: activate true/false
SK->>DP: initialize(plugin_config)
DP->>SP: _initialize_snapshot_publisher(kafka_bootstrap_servers)
SP->>KS: connect/start (best-effort)
KS-->>SP: ack / error
SP-->>DP: publisher created or disabled
SK->>DP: wire_handlers(registry, snapshot_publisher)
SK->>DP: wire_dispatchers(...)
SK->>DP: start_consumers(node_identity)
DP->>DP: form consumer-group using node_identity & purpose
Note over SK,DP: On shutdown (LIFO)
SK->>DP: shutdown()
DP->>SP: stop/cleanup (best-effort)
SP-->>DP: stopped/errored
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
No actionable comments were generated in the recent review. 🎉 Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/nodes/node_registration_orchestrator/plugin.py`:
- Around line 552-613: The try block in _initialize_snapshot_publisher creates
an AIOKafkaProducer and then awaits self._snapshot_publisher.start(); if start()
raises, the producer is leaked. Update _initialize_snapshot_publisher so that if
SnapshotPublisherRegistration.start() fails you explicitly stop/close the
underlying AIOKafkaProducer (await snapshot_producer.stop() or the appropriate
async close method) before abandoning the instance, and still set
self._snapshot_publisher = None; ensure the cleanup runs only when
snapshot_producer was created (e.g., by catching exceptions around the start
call or using a small try/except/finally scoped to the start step).
…lisher start failure
…nd config.name validation - Broadened except ImportError to also catch Exception when initializing PluginIntelligence, so an installed-but-broken dependency degrades gracefully instead of crashing the kernel - Added warning log when runtime_config.yaml is missing the 'name' field, so operators notice that plugin consumers will not subscribe Review iteration: 1/10
…gistration - Added ValueError/TypeError guard and port range check (1-65535) with fallback to 8500, matching the validation that existed in the old inline kernel wiring code Review iteration: 2/10
- Merged separate 'if not config.name' and 'if config.name' blocks into a single if/else to make the mutual exclusion explicit Review iteration: 3/10
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/service_kernel.py`:
- Around line 752-811: A plugin that succeeds in initialize(...) but later fails
in wire_handlers(...), wire_dispatchers(...), or start_consumers(...) can leave
open resources and is not added to activated_plugins for clean shutdown; add a
best-effort rollback path: after a failed wire_handlers, wire_dispatchers, or
start_consumers call, invoke a cleanup/teardown routine on that plugin (e.g.,
plugin.shutdown() or plugin.teardown() / plugin.dispose() — or call
plugin.start_consumers(plugin_config) rollback counterpart, then plugin.close/
plugin.cleanup if available), ensure any consumer unsubscribe callbacks created
so far are removed from plugin_unsubscribe_callbacks, and only append to
activated_plugins after all lifecycle steps succeed; update the activation loop
around plugin.initialize, plugin.wire_handlers, plugin.wire_dispatchers, and
plugin.start_consumers to perform this rollback on failure and log the cleanup
attempt.
Comment on lines
+752
to
811
| # Activate plugins in standard lifecycle order | ||
| plugin_activation_start = time.time() | ||
| for plugin in plugin_registry.get_all(): | ||
| plugin_id = plugin.plugin_id | ||
|
|
||
| # Initialize schema by executing SQL file directly | ||
| # Schema initialization is decoupled from the projector - it just ensures | ||
| # the table and indexes exist. The ProjectorShell uses the schema at runtime. | ||
| schema_file = ( | ||
| Path(__file__).parent.parent | ||
| / "schemas" | ||
| / "schema_registration_projection.sql" | ||
| # 1. Check activation | ||
| if not plugin.should_activate(plugin_config): | ||
| logger.info( | ||
| "Plugin '%s' skipped (not activated) (correlation_id=%s)", | ||
| plugin_id, | ||
| correlation_id, | ||
| ) | ||
| if schema_file.exists(): | ||
| try: | ||
| schema_sql = schema_file.read_text() | ||
| async with postgres_pool.acquire() as conn: | ||
| await conn.execute(schema_sql) | ||
| logger.info( | ||
| "Registration projection schema initialized (correlation_id=%s)", | ||
| correlation_id, | ||
| ) | ||
| except Exception as schema_error: | ||
| # Log warning but continue - schema may already exist | ||
| logger.warning( | ||
| "Schema initialization encountered error: %s (correlation_id=%s)", | ||
| sanitize_error_message(schema_error), | ||
| correlation_id, | ||
| extra={ | ||
| "error_type": type(schema_error).__name__, | ||
| }, | ||
| ) | ||
| else: | ||
| logger.warning( | ||
| "Schema file not found: %s (correlation_id=%s)", | ||
| schema_file, | ||
| correlation_id, | ||
| ) | ||
|
|
||
| # 4.6.5. Initialize HandlerConsul if Consul is configured | ||
| # CONSUL_HOST determines whether to enable Consul registration | ||
| consul_host = os.getenv("CONSUL_HOST") | ||
| if consul_host: | ||
| # Validate CONSUL_PORT environment variable | ||
| consul_port_str = os.getenv("CONSUL_PORT", "8500") | ||
| try: | ||
| consul_port = int(consul_port_str) | ||
| if not MIN_PORT <= consul_port <= MAX_PORT: | ||
| logger.warning( | ||
| "CONSUL_PORT %d outside valid range %d-%d, using default 8500 (correlation_id=%s)", | ||
| consul_port, | ||
| MIN_PORT, | ||
| MAX_PORT, | ||
| correlation_id, | ||
| ) | ||
| consul_port = 8500 | ||
| except ValueError: | ||
| logger.warning( | ||
| "Invalid CONSUL_PORT value '%s', using default 8500 (correlation_id=%s)", | ||
| consul_port_str, | ||
| correlation_id, | ||
| ) | ||
| consul_port = 8500 | ||
|
|
||
| try: | ||
| # Deferred import: Only load HandlerConsul when Consul is configured. | ||
| # This avoids loading the consul dependency (and its transitive deps) | ||
| # when Consul integration is disabled, reducing startup time. | ||
| from omnibase_infra.handlers import HandlerConsul | ||
|
|
||
| consul_handler = HandlerConsul(container) | ||
| await consul_handler.initialize( | ||
| { | ||
| "host": consul_host, | ||
| "port": consul_port, | ||
| } | ||
| ) | ||
| logger.info( | ||
| "HandlerConsul initialized for dual registration (correlation_id=%s)", | ||
| correlation_id, | ||
| extra={ | ||
| "consul_host": consul_host, | ||
| "consul_port": consul_port, | ||
| }, | ||
| ) | ||
| except Exception as consul_error: | ||
| # Log warning but continue without Consul (PostgreSQL is source of truth) | ||
| # Use sanitize_error_message to prevent credential leakage in logs | ||
| logger.warning( | ||
| "Failed to initialize HandlerConsul, proceeding without Consul: %s (correlation_id=%s)", | ||
| sanitize_error_message(consul_error), | ||
| correlation_id, | ||
| extra={ | ||
| "error_type": type(consul_error).__name__, | ||
| }, | ||
| ) | ||
| consul_handler = None | ||
| else: | ||
| logger.debug( | ||
| "CONSUL_HOST not set, Consul registration disabled (correlation_id=%s)", | ||
| correlation_id, | ||
| ) | ||
|
|
||
| # 4.6.6. Create SnapshotPublisherRegistration if Kafka is available | ||
| # Snapshot publishing is best-effort - if creation or start fails, | ||
| # the system continues without snapshot publishing. | ||
| snapshot_publisher = None | ||
| if kafka_bootstrap_servers: | ||
| try: | ||
| from aiokafka import AIOKafkaProducer | ||
|
|
||
| from omnibase_infra.models.projection import ( | ||
| ModelSnapshotTopicConfig, | ||
| ) | ||
| from omnibase_infra.projectors.snapshot_publisher_registration import ( | ||
| SnapshotPublisherRegistration, | ||
| ) | ||
| continue | ||
|
|
||
| snapshot_config = ModelSnapshotTopicConfig.default() | ||
| snapshot_producer = AIOKafkaProducer( | ||
| bootstrap_servers=kafka_bootstrap_servers, | ||
| ) | ||
| snapshot_publisher = SnapshotPublisherRegistration( | ||
| snapshot_producer, | ||
| snapshot_config, | ||
| bootstrap_servers=kafka_bootstrap_servers, | ||
| ) | ||
| await snapshot_publisher.start() | ||
| logger.info( | ||
| "SnapshotPublisherRegistration started for topic %s (correlation_id=%s)", | ||
| snapshot_config.topic, | ||
| correlation_id, | ||
| extra={ | ||
| "topic": snapshot_config.topic, | ||
| "bootstrap_servers": kafka_bootstrap_servers, | ||
| }, | ||
| ) | ||
| except Exception as snap_pub_error: | ||
| # Log warning but continue without snapshot publishing | ||
| logger.warning( | ||
| "Failed to start SnapshotPublisherRegistration, " | ||
| "continuing without snapshot publishing: %s (correlation_id=%s)", | ||
| sanitize_error_message(snap_pub_error), | ||
| correlation_id, | ||
| extra={ | ||
| "error_type": type(snap_pub_error).__name__, | ||
| }, | ||
| ) | ||
| snapshot_publisher = None | ||
| else: | ||
| logger.debug( | ||
| "KAFKA_BOOTSTRAP_SERVERS not set, snapshot publishing disabled (correlation_id=%s)", | ||
| correlation_id, | ||
| ) | ||
|
|
||
| # 4.7. Wire registration handlers with projector and consul_handler | ||
| registration_summary = await wire_registration_handlers( | ||
| container, | ||
| postgres_pool, | ||
| projector=projector, | ||
| consul_handler=consul_handler, | ||
| snapshot_publisher=snapshot_publisher, | ||
| # 2. Initialize (create pools, connections, resources) | ||
| init_result = await plugin.initialize(plugin_config) | ||
| if not init_result: | ||
| logger.warning( | ||
| "Plugin '%s' initialization failed: %s (correlation_id=%s)", | ||
| plugin_id, | ||
| init_result.get_error_message_or_default(), | ||
| correlation_id, | ||
| ) | ||
| logger.info( | ||
| "Registration handlers wired (correlation_id=%s)", | ||
| continue | ||
|
|
||
| # 3. Wire handlers | ||
| wire_result = await plugin.wire_handlers(plugin_config) | ||
| if not wire_result: | ||
| logger.warning( | ||
| "Plugin '%s' handler wiring failed: %s (correlation_id=%s)", | ||
| plugin_id, | ||
| wire_result.get_error_message_or_default(), | ||
| correlation_id, | ||
| extra={ | ||
| "services": registration_summary["services"], | ||
| }, | ||
| ) | ||
| continue | ||
|
|
||
| # 4.8. Create introspection dispatcher for routing events | ||
| # Deferred import: HandlerNodeIntrospected depends on PostgreSQL and | ||
| # registration infrastructure. Only loaded after PostgreSQL pool is | ||
| # successfully created and registration handlers are wired. | ||
| from omnibase_infra.nodes.node_registration_orchestrator.handlers import ( | ||
| HandlerNodeIntrospected, | ||
| # 4. Wire dispatchers (non-fatal if skipped) | ||
| dispatch_result = await plugin.wire_dispatchers(plugin_config) | ||
| if not dispatch_result: | ||
| logger.warning( | ||
| "Plugin '%s' dispatcher wiring failed: %s (correlation_id=%s)", | ||
| plugin_id, | ||
| dispatch_result.get_error_message_or_default(), | ||
| correlation_id, | ||
| ) | ||
|
|
||
| # Check if service_registry is available (may be None in omnibase_core 0.6.x) | ||
| if container.service_registry is None: | ||
| logger.warning( | ||
| "DEGRADED_MODE: ServiceRegistry not available, skipping introspection dispatcher creation (correlation_id=%s)", | ||
| correlation_id, | ||
| extra={ | ||
| "error_type": "NoneType", | ||
| "correlation_id": correlation_id, | ||
| "degraded_mode": True, | ||
| "degraded_reason": "service_registry_unavailable", | ||
| "component": "introspection_dispatcher", | ||
| }, | ||
| ) | ||
| # Set introspection_dispatcher to None and continue without it | ||
| introspection_dispatcher = None | ||
| else: | ||
| logger.debug( | ||
| "Resolving HandlerNodeIntrospected from container (correlation_id=%s)", | ||
| correlation_id, | ||
| ) | ||
| handler_introspected: HandlerNodeIntrospected = ( | ||
| await container.service_registry.resolve_service( | ||
| HandlerNodeIntrospected | ||
| ) | ||
| ) | ||
| logger.debug( | ||
| "HandlerNodeIntrospected resolved successfully (correlation_id=%s)", | ||
| correlation_id, | ||
| extra={ | ||
| "handler_class": handler_introspected.__class__.__name__, | ||
| }, | ||
| ) | ||
| # 5. Start consumers | ||
| consumer_result = await plugin.start_consumers(plugin_config) | ||
| if consumer_result and consumer_result.unsubscribe_callbacks: | ||
| plugin_unsubscribe_callbacks.extend( | ||
| consumer_result.unsubscribe_callbacks | ||
| ) | ||
|
|
||
| introspection_dispatcher = DispatcherNodeIntrospected( | ||
| handler_introspected | ||
| ) | ||
| logger.info( | ||
| "Introspection dispatcher created and wired (correlation_id=%s)", | ||
| correlation_id, | ||
| extra={ | ||
| "dispatcher_class": introspection_dispatcher.__class__.__name__, | ||
| "handler_class": handler_introspected.__class__.__name__, | ||
| }, | ||
| ) | ||
| activated_plugins.append(plugin) | ||
| logger.info( | ||
| "Plugin '%s' activated successfully (correlation_id=%s)", | ||
| plugin_id, | ||
| correlation_id, | ||
| ) | ||
|
|
There was a problem hiding this comment.
Clean up partially activated plugins to avoid resource leaks.
If a plugin initializes successfully but fails at wire_handlers or start_consumers, its resources (pools/producers/etc.) remain live and the plugin isn’t tracked for shutdown. Add best‑effort rollback before continuing.
Suggested fix (best-effort cleanup on partial activation)
wire_result = await plugin.wire_handlers(plugin_config)
if not wire_result:
logger.warning(
"Plugin '%s' handler wiring failed: %s (correlation_id=%s)",
plugin_id,
wire_result.get_error_message_or_default(),
correlation_id,
)
+ try:
+ await plugin.shutdown(plugin_config)
+ except Exception as cleanup_error:
+ logger.warning(
+ "Plugin '%s' cleanup after handler wiring failure failed: %s (correlation_id=%s)",
+ plugin_id,
+ sanitize_error_message(cleanup_error),
+ correlation_id,
+ )
continue
@@
consumer_result = await plugin.start_consumers(plugin_config)
+ if not consumer_result:
+ logger.warning(
+ "Plugin '%s' consumer startup failed: %s (correlation_id=%s)",
+ plugin_id,
+ consumer_result.get_error_message_or_default(),
+ correlation_id,
+ )
+ try:
+ await plugin.shutdown(plugin_config)
+ except Exception as cleanup_error:
+ logger.warning(
+ "Plugin '%s' cleanup after consumer startup failure failed: %s (correlation_id=%s)",
+ plugin_id,
+ sanitize_error_message(cleanup_error),
+ correlation_id,
+ )
+ continue
if consumer_result and consumer_result.unsubscribe_callbacks:
plugin_unsubscribe_callbacks.extend(
consumer_result.unsubscribe_callbacks
)🤖 Prompt for AI Agents
In `@src/omnibase_infra/runtime/service_kernel.py` around lines 752 - 811, A
plugin that succeeds in initialize(...) but later fails in wire_handlers(...),
wire_dispatchers(...), or start_consumers(...) can leave open resources and is
not added to activated_plugins for clean shutdown; add a best-effort rollback
path: after a failed wire_handlers, wire_dispatchers, or start_consumers call,
invoke a cleanup/teardown routine on that plugin (e.g., plugin.shutdown() or
plugin.teardown() / plugin.dispose() — or call
plugin.start_consumers(plugin_config) rollback counterpart, then plugin.close/
plugin.cleanup if available), ensure any consumer unsubscribe callbacks created
so far are removed from plugin_unsubscribe_callbacks, and only append to
activated_plugins after all lifecycle steps succeed; update the activation loop
around plugin.initialize, plugin.wire_handlers, plugin.wire_dispatchers, and
plugin.start_consumers to perform this rollback on failure and log the cleanup
attempt.
…xceptions Wrap per-plugin activation block in try/except so that an unexpected exception from an external plugin (e.g. PluginIntelligence) does not crash the entire kernel. Matches the graceful degradation intent already established by the constructor guard. Review iteration: 1/10
Track plugin for shutdown immediately after successful initialize() so allocated resources (PostgreSQL pools, Kafka producers) are always cleaned up even if subsequent lifecycle steps (wire_handlers, wire_dispatchers, start_consumers) fail. Review iteration: 2/10
…scalation - Added safety-net shutdown for plugins that fail before being tracked in activated_plugins, preventing potential resource leaks from partially-initialized plugins during lifecycle activation. - Escalated missing config.name log from warning to error level so operators notice when introspection consumers are silently disabled. Review iteration: 1/10
The safety-net shutdown for untracked plugins was missing the required config argument, which would raise TypeError at runtime and defeat the resource leak prevention. Review iteration: 2/10
jonahgabriel
deleted the
jonah/omn-1992-activate-pluginintelligence-in-kernel-bootstrap-via
branch
February 8, 2026 20:38
2 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
service_kernel.pywith a plugin registry loop calling the standardProtocolDomainPluginlifecyclePluginRegistrationandPluginIntelligencewithRegistryDomainPluginduring kernel initPluginRegistrationlifecycle methods (initialize,wire_handlers,wire_dispatchers,start_consumers,shutdown)node_identityandkafka_bootstrap_serversfields toModelDomainPluginConfigomniintelligenceif not installed (no import errors)Net reduction: -261 lines in kernel bootstrap (418 insertions, 679 deletions across 3 files)
Test plan
Summary by CodeRabbit
New Features
Improvements
Refactor