feat(chain-learning): prompt-chain learning system with 5 ONEX nodes [OMN-7032]

src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py

@@ -18,11 +18,18 @@ class EnumOmnibaseInfraTopic(str, Enum):
     All values are raw topic strings as declared in contract.yaml.
     Members are sorted by (kind, event_name, version).
     """
+    CMD_CHAIN_LEARN_V1 = "onex.cmd.omnibase-infra.chain-learn.v1"  # onex.cmd.omnibase-infra.chain-learn.v1
     CMD_CONSUMER_RESTART_V1 = "onex.cmd.omnibase-infra.consumer-restart.v1"  # onex.cmd.omnibase-infra.consumer-restart.v1
     CMD_DELEGATION_REQUEST_V1 = "onex.cmd.omnibase-infra.delegation-request.v1"  # onex.cmd.omnibase-infra.delegation-request.v1
     CMD_LLM_EMBEDDING_REQUEST_V1 = "onex.cmd.omnibase-infra.llm-embedding-request.v1"  # onex.cmd.omnibase-infra.llm-embedding-request.v1
     CMD_LLM_INFERENCE_REQUEST_V1 = "onex.cmd.omnibase-infra.llm-inference-request.v1"  # onex.cmd.omnibase-infra.llm-inference-request.v1
     EVT_BASELINES_COMPUTED_V1 = "onex.evt.omnibase-infra.baselines-computed.v1"  # onex.evt.omnibase-infra.baselines-computed.v1
+    EVT_CHAIN_LEARN_COMPLETE_V1 = "onex.evt.omnibase-infra.chain-learn-complete.v1"  # onex.evt.omnibase-infra.chain-learn-complete.v1
+    EVT_CHAIN_LEARN_FAILED_V1 = "onex.evt.omnibase-infra.chain-learn-failed.v1"  # onex.evt.omnibase-infra.chain-learn-failed.v1
+    EVT_CHAIN_REPLAY_RESULT_V1 = "onex.evt.omnibase-infra.chain-replay-result.v1"  # onex.evt.omnibase-infra.chain-replay-result.v1
+    EVT_CHAIN_RETRIEVAL_RESULT_V1 = "onex.evt.omnibase-infra.chain-retrieval-result.v1"  # onex.evt.omnibase-infra.chain-retrieval-result.v1
+    EVT_CHAIN_STORED_V1 = "onex.evt.omnibase-infra.chain-stored.v1"  # onex.evt.omnibase-infra.chain-stored.v1
+    EVT_CHAIN_VERIFIED_V1 = "onex.evt.omnibase-infra.chain-verified.v1"  # onex.evt.omnibase-infra.chain-verified.v1
     EVT_CONSUMER_HEALTH_V1 = "onex.evt.omnibase-infra.consumer-health.v1"  # onex.evt.omnibase-infra.consumer-health.v1
     EVT_DB_ERROR_V1 = "onex.evt.omnibase-infra.db-error.v1"  # onex.evt.omnibase-infra.db-error.v1
     EVT_DELEGATION_COMPLETED_V1 = "onex.evt.omnibase-infra.delegation-completed.v1"  # onex.evt.omnibase-infra.delegation-completed.v1

src/omnibase_infra/nodes/node_chain_orchestrator/__init__.py

@@ -0,0 +1,7 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Chain learning orchestrator node."""
+
+from .node import NodeChainOrchestrator
+
+__all__ = ["NodeChainOrchestrator"]

src/omnibase_infra/nodes/node_chain_orchestrator/contract.yaml

@@ -0,0 +1,115 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+# ONEX Node Contract - Chain Learning Orchestrator
+#
+# Top-level coordinator for the prompt-chain learning workflow.
+# Manages: retrieval -> decide shortcut/explore -> execute -> verify -> store.
+#
+name: "node_chain_orchestrator"
+node_type: "ORCHESTRATOR_GENERIC"
+contract_version:
+  major: 1
+  minor: 0
+  patch: 0
+node_version: "1.0.0"
+description: >
+  Orchestrates the prompt-chain learning workflow: receive command, dispatch chain retrieval, evaluate hit/miss, dispatch replay or explore, verify contract, store verified chain.
+
+event_bus:
+  version:
+    major: 1
+    minor: 0
+    patch: 0
+  subscribe_topics:
+    - "onex.cmd.omnibase-infra.chain-learn.v1"
+    - "onex.evt.omnibase-infra.chain-retrieval-result.v1"
+    - "onex.evt.omnibase-infra.chain-replay-result.v1"
+    - "onex.evt.omnibase-infra.chain-verified.v1"
+    - "onex.evt.omnibase-infra.chain-stored.v1"
+  publish_topics:
+    - "onex.evt.omnibase-infra.chain-learn-complete.v1"
+    - "onex.evt.omnibase-infra.chain-learn-failed.v1"
+input_model:
+  name: "ModelChainLearnCommand"
+  module: "omnibase_infra.nodes.node_chain_orchestrator.models.model_chain_learn_command"
+  description: "Command to initiate chain learning workflow."
+output_model:
+  name: "ModelChainLearnResult"
+  module: "omnibase_infra.nodes.node_chain_orchestrator.models.model_chain_learn_result"
+  description: "Final chain learning result with path taken and timing."
+handler_routing:
+  routing_strategy: "payload_type_match"
+  handlers:
+    - event_model:
+        name: "ModelChainLearnCommand"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.models.model_chain_learn_command"
+      handler:
+        name: "HandlerChainLearnInitiate"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.handlers.handler_chain_learn_initiate"
+      output_events:
+        - "ModelChainLearnCommand"
+    - event_model:
+        name: "ModelChainRetrievalResult"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.models.model_chain_retrieval_result"
+      handler:
+        name: "HandlerChainRetrievalComplete"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.handlers.handler_chain_retrieval_complete"
+      output_events:
+        - "ModelChainReplayInput"
+    - event_model:
+        name: "ModelChainReplayResult"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.models.model_chain_replay_result"
+      handler:
+        name: "HandlerChainReplayComplete"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.handlers.handler_chain_replay_complete"
+      output_events:
+        - "ModelChainStoreRequest"
+    - event_model:
+        name: "ModelChainStoreResult"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.models.model_chain_store_result"
+      handler:
+        name: "HandlerChainStoreComplete"
+        module: "omnibase_infra.nodes.node_chain_orchestrator.handlers.handler_chain_store_complete"
+      output_events:
+        - "ModelChainLearnResult"
+workflow_coordination:
+  workflow_definition:
+    workflow_metadata:
+      workflow_name: "chain_learning_workflow"
+      workflow_version:
+        major: 1
+        minor: 0
+        patch: 0
+      description: "Prompt-chain learning: cache verified trajectories for fast replay"
+    execution_graph:
+      nodes:
+        - node_id: "retrieve"
+          node_type: EFFECT_GENERIC
+          description: "Embed prompt and search Qdrant for similar chains"
+        - node_id: "replay"
+          node_type: COMPUTE_GENERIC
+          description: "Adapt cached chain to new context (HIT path only)"
+          depends_on: ["retrieve"]
+        - node_id: "verify"
+          node_type: REDUCER_GENERIC
+          description: "FSM tracking verification state"
+        - node_id: "store"
+          node_type: EFFECT_GENERIC
+          description: "Store verified chain to Qdrant"
+          depends_on: ["verify"]
+    coordination_rules:
+      execution_mode: sequential
+      parallel_execution_allowed: false
+      max_parallel_branches: 1
+      failure_recovery_strategy: fail_fast
+      max_retries: 0
+      timeout_ms: 30000
+metadata:
+  description: "Chain learning orchestrator -- cache and replay verified workflows"
+  author: "OmniNode Team"
+  created: "2026-04-02"
+  tags:
+    - "orchestrator"
+    - "chain-learning"
+    - "cache"
+    - "trajectory"

src/omnibase_infra/nodes/node_chain_orchestrator/handlers/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Chain orchestrator handlers."""

src/omnibase_infra/nodes/node_chain_orchestrator/handlers/handler_chain_learn_initiate.py

@@ -0,0 +1,59 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Handler that initiates the chain learning workflow.
+
+Receives the inbound command and dispatches chain retrieval.
+"""
+
+from __future__ import annotations
+
+import logging
+from uuid import UUID
+
+from omnibase_infra.enums import EnumHandlerType, EnumHandlerTypeCategory
+
+logger = logging.getLogger(__name__)
+
+
+class HandlerChainLearnInitiate:
+    """Receives chain learn command and dispatches retrieval."""
+
+    @property
+    def handler_type(self) -> EnumHandlerType:
+        return EnumHandlerType.NODE_HANDLER
+
+    @property
+    def handler_category(self) -> EnumHandlerTypeCategory:
+        return EnumHandlerTypeCategory.COMPUTE
+
+    async def handle(
+        self,
+        prompt_text: str,
+        workflow_ref: str,
+        correlation_id: UUID,
+        similarity_threshold: float = 0.85,
+    ) -> dict[str, str | float]:
+        """Validate the inbound command and prepare retrieval dispatch.
+
+        Args:
+            prompt_text: The prompt to learn from.
+            workflow_ref: The workflow that produced the chain.
+            correlation_id: Workflow correlation ID.
+            similarity_threshold: Minimum similarity for cache hit.
+
+        Returns:
+            Dict with dispatch parameters for the retrieval effect.
+        """
+        logger.info(
+            "Chain learn initiated: workflow=%s, threshold=%.2f (correlation_id=%s)",
+            workflow_ref,
+            similarity_threshold,
+            correlation_id,
+        )
+
+        return {
+            "prompt_text": prompt_text,
+            "workflow_ref": workflow_ref,
+            "correlation_id": str(correlation_id),
+            "similarity_threshold": str(similarity_threshold),
+        }

src/omnibase_infra/nodes/node_chain_orchestrator/handlers/handler_chain_replay_complete.py

@@ -0,0 +1,69 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Handler that processes replay results and dispatches verification."""
+
+from __future__ import annotations
+
+import logging
+from uuid import UUID
+
+from omnibase_infra.enums import EnumHandlerType, EnumHandlerTypeCategory
+from omnibase_infra.nodes.node_chain_orchestrator.models import (
+    ModelChainReplayResult,
+)
+
+logger = logging.getLogger(__name__)
+
+# Minimum confidence to proceed with replay (below this -> fallback)
+MIN_REPLAY_CONFIDENCE = 0.6
+
+
+class HandlerChainReplayComplete:
+    """Evaluates replay result and dispatches store or fallback."""
+
+    @property
+    def handler_type(self) -> EnumHandlerType:
+        return EnumHandlerType.NODE_HANDLER
+
+    @property
+    def handler_category(self) -> EnumHandlerTypeCategory:
+        return EnumHandlerTypeCategory.COMPUTE
+
+    async def handle(
+        self,
+        replay_result: ModelChainReplayResult,
+        correlation_id: UUID,
+    ) -> dict[str, str | float]:
+        """Evaluate replay result and decide next step.
+
+        Args:
+            replay_result: Result from chain replay compute.
+            correlation_id: Workflow correlation ID.
+
+        Returns:
+            Dict with decision: proceed to store or fallback to explore.
+        """
+        if replay_result.confidence >= MIN_REPLAY_CONFIDENCE:
+            logger.info(
+                "Replay accepted: confidence=%.2f, steps=%d (correlation_id=%s)",
+                replay_result.confidence,
+                len(replay_result.adapted_steps),
+                correlation_id,
+            )
+            return {
+                "action": "verify",
+                "confidence": str(replay_result.confidence),
+                "summary": replay_result.adaptation_summary,
+            }
+
+        logger.info(
+            "Replay rejected: confidence=%.2f < %.2f (correlation_id=%s)",
+            replay_result.confidence,
+            MIN_REPLAY_CONFIDENCE,
+            correlation_id,
+        )
+        return {
+            "action": "fallback",
+            "confidence": str(replay_result.confidence),
+            "reason": f"Confidence {replay_result.confidence:.2f} below threshold {MIN_REPLAY_CONFIDENCE:.2f}",
+        }

src/omnibase_infra/nodes/node_chain_orchestrator/handlers/handler_chain_retrieval_complete.py

@@ -0,0 +1,65 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""Handler that evaluates retrieval results and dispatches replay or explore."""
+
+from __future__ import annotations
+
+import logging
+from uuid import UUID
+
+from omnibase_infra.enums import EnumHandlerType, EnumHandlerTypeCategory
+from omnibase_infra.nodes.node_chain_orchestrator.models import (
+    ModelChainRetrievalResult,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class HandlerChainRetrievalComplete:
+    """Evaluates hit/miss from retrieval and decides next step."""
+
+    @property
+    def handler_type(self) -> EnumHandlerType:
+        return EnumHandlerType.NODE_HANDLER
+
+    @property
+    def handler_category(self) -> EnumHandlerTypeCategory:
+        return EnumHandlerTypeCategory.COMPUTE
+
+    async def handle(
+        self,
+        retrieval_result: ModelChainRetrievalResult,
+        correlation_id: UUID,
+    ) -> dict[str, str | bool]:
+        """Evaluate retrieval result and decide path.
+
+        Args:
+            retrieval_result: Result from chain retrieval effect.
+            correlation_id: Workflow correlation ID.
+
+        Returns:
+            Dict with path decision and data for next step.
+        """
+        if retrieval_result.is_hit and retrieval_result.matches:
+            best_match = retrieval_result.matches[0]
+            logger.info(
+                "Chain HIT: similarity=%.3f, chain_id=%s (correlation_id=%s)",
+                best_match.similarity_score,
+                best_match.chain_entry.chain_id,
+                correlation_id,
+            )
+            return {
+                "path": "replay",
+                "chain_id": str(best_match.chain_entry.chain_id),
+                "is_hit": True,
+            }
+
+        logger.info(
+            "Chain MISS: best_similarity=%.3f (correlation_id=%s)",
+            retrieval_result.best_match_similarity,
+            correlation_id,
+        )
+        return {
+            "path": "explore",
+            "is_hit": False,
+        }