feat(OMN-12132): add node_log_persistence_effect EFFECT node

pyproject.toml

@@ -116,6 +116,7 @@ node_golden_chain_sweep = "omnimarket.nodes.node_golden_chain_sweep"
 node_golden_chain_generator = "omnimarket.nodes.node_golden_chain_generator"
 node_runtime_sweep = "omnimarket.nodes.node_runtime_sweep"
 node_log_projection = "omnimarket.nodes.node_log_projection"
+node_log_persistence_effect = "omnimarket.nodes.node_log_persistence_effect"
 node_data_verification = "omnimarket.nodes.node_data_verification"
 node_process_watchdog = "omnimarket.nodes.node_process_watchdog"
 node_projection_session_outcome = "omnimarket.nodes.node_projection_session_outcome"

src/omnimarket/nodes/node_log_persistence_effect/__init__.py

@@ -0,0 +1,11 @@
+"""node_log_persistence_effect — Persists structured log events to Postgres."""
+
+from omnimarket.nodes.node_log_persistence_effect.handlers.handler_log_persistence_effect import (
+    ModelLogPersistenceResult,
+    NodeLogPersistenceEffect,
+)
+
+__all__ = [
+    "ModelLogPersistenceResult",
+    "NodeLogPersistenceEffect",
+]

src/omnimarket/nodes/node_log_persistence_effect/contract.yaml

@@ -0,0 +1,38 @@
+---
+name: log_persistence_effect
+contract_version: {major: 1, minor: 0, patch: 0}
+node_type: effect
+node_version: {major: 1, minor: 0, patch: 0}
+description: >
+  Persists structured log events to Postgres for historical trace queries.
+  Enables the trace explorer to query beyond live Kafka retention window.
+
+terminal_event: onex.evt.omnimarket.log-persistence-completed.v1
+
+handler:
+  module: omnimarket.nodes.node_log_persistence_effect.handlers.handler_log_persistence_effect
+  class: NodeLogPersistenceEffect
+
+handler_routing:
+  routing_strategy: operation_match
+  handlers:
+    - handler:
+        name: NodeLogPersistenceEffect
+        module: omnimarket.nodes.node_log_persistence_effect.handlers.handler_log_persistence_effect
+
+event_bus:
+  subscribe_topics:
+    - onex.evt.platform.log-entry.v1
+  publish_topics:
+    - onex.evt.omnimarket.log-persistence-completed.v1
+
+descriptor:
+  node_archetype: effect
+  purity: effectful
+  idempotent: true
+  timeout_ms: 10000
+  runtime_profiles:
+    - effects
+
+metadata:
+  transport_type: kafka

src/omnimarket/nodes/node_log_persistence_effect/handlers/__init__.py

@@ -0,0 +1 @@
+"""Handlers for node_log_persistence_effect."""

src/omnimarket/nodes/node_log_persistence_effect/handlers/handler_log_persistence_effect.py

@@ -0,0 +1,179 @@
+# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+
+"""NodeLogPersistenceEffect — persists log events to Postgres.
+
+Subscribes to onex.evt.platform.log-entry.v1 events and INSERTs each into
+the log_entries table. Idempotent on entry_id via ON CONFLICT DO NOTHING.
+
+If the asyncpg pool is None (no DB configured), the handler logs a warning
+and skips the write rather than raising — graceful degradation for
+environments without Postgres.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Literal
+from uuid import uuid4
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from omnimarket.nodes.contract_topics import (
+    contract_publish_topics,
+    contract_subscribe_topics,
+)
+from omnimarket.nodes.node_log_projection.handlers.handler_log_projection import (
+    ModelLogEntry,
+)
+
+if TYPE_CHECKING:
+    import asyncpg
+
+logger = logging.getLogger(__name__)
+
+_CONTRACT_PATH = Path(__file__).parent.parent / "contract.yaml"
+_SUBSCRIBE_TOPICS = contract_subscribe_topics(_CONTRACT_PATH)
+_PUBLISH_TOPICS = contract_publish_topics(_CONTRACT_PATH)
+
+_DEFAULT_PG_DSN = os.environ.get("ONEX_PG_DSN", "")  # contract-config-ok: config  # fmt: skip
+
+
+class ModelLogPersistenceResult(BaseModel):
+    """Result of a single log entry persistence attempt."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    result_id: str = Field(default_factory=lambda: str(uuid4()))
+    entry_id: str = Field(..., description="entry_id of the persisted log entry.")
+    status: Literal["written", "skipped", "idempotent", "error"] = Field(...)
+    error_message: str | None = Field(default=None)
+
+
+class NodeLogPersistenceEffect:
+    """EFFECT node: persists log events from Kafka to Postgres.
+
+    Accepts an optional asyncpg pool for dependency injection (enables unit
+    testing without a real database). When pool is None and no DSN is
+    configured, operates in no-op mode with a logged warning.
+    """
+
+    handler_type: Literal["node_handler"] = "node_handler"
+    handler_category: Literal["effect"] = "effect"
+
+    def __init__(
+        self,
+        pool: asyncpg.Pool | None = None,
+        pg_dsn: str = _DEFAULT_PG_DSN,
+    ) -> None:
+        self._injected_pool = pool
+        self._pg_dsn = pg_dsn
+        self._pool: asyncpg.Pool | None = pool
+
+    async def _get_pool(self) -> asyncpg.Pool | None:
+        """Return pool, lazily creating it from DSN if not injected."""
+        if self._pool is not None:
+            return self._pool
+        if not self._pg_dsn:
+            return None
+        import asyncpg as _asyncpg
+
+        self._pool = await _asyncpg.create_pool(
+            dsn=self._pg_dsn,
+            min_size=1,
+            max_size=5,
+            command_timeout=10,
+        )
+        return self._pool
+
+    async def handle(self, entry: ModelLogEntry) -> ModelLogPersistenceResult:
+        """Persist a single log entry to Postgres.
+
+        Args:
+            entry: The structured log event to persist.
+
+        Returns:
+            ModelLogPersistenceResult describing the outcome.
+        """
+        pool = await self._get_pool()
+
+        if pool is None:
+            logger.warning(
+                "NodeLogPersistenceEffect: no DB pool available, skipping entry_id=%s",
+                entry.entry_id,
+            )
+            return ModelLogPersistenceResult(
+                entry_id=entry.entry_id,
+                status="skipped",
+            )
+
+        try:
+            inserted = await self._insert(pool, entry)
+            status: Literal["written", "idempotent"] = (
+                "written" if inserted else "idempotent"
+            )
+            logger.debug(
+                "NodeLogPersistenceEffect: entry_id=%s status=%s",
+                entry.entry_id,
+                status,
+            )
+            return ModelLogPersistenceResult(entry_id=entry.entry_id, status=status)
+        except Exception as exc:
+            logger.error(
+                "NodeLogPersistenceEffect: failed to persist entry_id=%s: %s",
+                entry.entry_id,
+                exc,
+            )
+            return ModelLogPersistenceResult(
+                entry_id=entry.entry_id,
+                status="error",
+                error_message=str(exc),
+            )
+
+    async def _insert(self, pool: asyncpg.Pool, entry: ModelLogEntry) -> bool:
+        """INSERT entry into log_entries. Returns True if a row was inserted."""
+        metadata_json = json.dumps(entry.metadata)
+        async with pool.acquire() as conn:
+            inserted_id = await conn.fetchval(
+                """
+                INSERT INTO log_entries (
+                    entry_id,
+                    timestamp,
+                    node_name,
+                    function_name,
+                    level,
+                    message,
+                    correlation_id,
+                    duration_ms,
+                    metadata
+                ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+                ON CONFLICT (entry_id) DO NOTHING
+                RETURNING entry_id
+                """,
+                entry.entry_id,
+                entry.timestamp,
+                entry.node_name,
+                entry.function_name,
+                entry.level.value,
+                entry.message,
+                entry.correlation_id,
+                entry.duration_ms,
+                metadata_json,
+            )
+        return inserted_id is not None
+
+    @staticmethod
+    def handle_raw(input_data: dict[str, Any]) -> dict[str, Any]:
+        """Synchronous dict-in / dict-out shim for runtime protocol compatibility."""
+        entry = ModelLogEntry(**input_data)
+        return {
+            "entry_id": entry.entry_id,
+            "status": "skipped",
+            "error_message": "use async handle() for live persistence",
+        }
+
+
+__all__: list[str] = ["ModelLogPersistenceResult", "NodeLogPersistenceEffect"]

src/omnimarket/nodes/node_log_persistence_effect/metadata.yaml

@@ -0,0 +1,24 @@
+name: node_log_persistence_effect
+version: "1.0.0"
+description: "Persists structured log events to Postgres for historical trace queries"
+omnibase_core_compat: ">=0.39.0,<1.0.0"
+entry_points:
+  onex.nodes:
+    node_log_persistence_effect: "omnimarket.nodes.node_log_persistence_effect"
+capabilities:
+  standalone: true
+  full_runtime: true
+  requires_network: false
+  requires_repo: false
+  requires_secrets: false
+  requires_docker: false
+  side_effect_class: external_io
+dependencies:
+  - "omnibase_core>=0.39.0"
+  - "asyncpg>=0.29.0"
+authors: ["OmniNode Platform Team"]
+license: "MIT"
+tags: ["effect", "log-persistence", "observability", "postgres"]
+pack: "data"
+display_name: "Log Persistence Effect"
+node_role: "effect"

tests/unit/nodes/test_node_log_persistence_effect/__init__.py

@@ -0,0 +1 @@
+"""Unit tests for node_log_persistence_effect."""