Repository navigation
feat(OMN-1908): Add cross-repo validation ledger persistence - #257
Merged
jonahgabriel merged 10 commits intoFeb 7, 2026
Merged
Conversation
Add append-only validation event ledger for cross-repo validation persistence. This enables tracking validation runs, violations, and completions across repository boundaries with full replay capability. Components: - PostgreSQL migration for validation_event_ledger table with partitioning support and idempotent append (content-hash dedup) - Pydantic models: entry, query, replay batch, append result - Repository protocol and Postgres implementation with flexible querying, TTL-based cleanup, and Kafka position tracking - COMPUTE node with handler for projecting Kafka validation events into ledger entries (base64 encoding, SHA-256 hashing) - 75 unit tests covering models, handler, and repository
📝 WalkthroughWalkthroughAdds validation-event DB migrations (create with pgcrypto and destructive DROP rollback), a DI registry for validation ledger projection, stricter envelope/base64 and envelope_hash validation, handler error/context refactors to RuntimeHostError with correlation_id propagation, prepends kafka_topic to deterministic ordering, and test updates. Changes
Sequence Diagram(s)sequenceDiagram
participant Kafka as Kafka (topic)
participant Handler as LedgerProjectionHandler
participant DI as DI Registry / Container
participant DB as Postgres ValidationLedger
Kafka->>Handler: deliver message (headers, value/base64)
Handler->>DI: resolve NodeValidationLedgerProjectionCompute
Handler->>Handler: extract metadata (correlation_id from headers, timestamp -> UTC)
alt invalid or empty envelope/value
Handler-->>Kafka: raise RuntimeHostError (ModelInfraErrorContext, transport=KAFKA, correlation_id)
else valid envelope
Handler->>DB: persist/read projection (ORDER BY kafka_topic,kafka_partition,kafka_offset)
DB-->>Handler: ack/result
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 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)
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: 6
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/migrations/002_create_validation_event_ledger.sql`:
- Around line 41-44: Migration uses gen_random_uuid() in the
validation_event_ledger table definition (id UUID PRIMARY KEY DEFAULT
gen_random_uuid()), but the pgcrypto extension is not enabled; add a step to
initialize pgcrypto before any gen_random_uuid() usage by adding "CREATE
EXTENSION IF NOT EXISTS pgcrypto;" at the start of the migration that first uses
gen_random_uuid() (e.g., include it in this migration or the prior migration
that introduces event_ledger) so the DEFAULT gen_random_uuid() call succeeds on
fresh databases.
In
`@src/omnibase_infra/nodes/node_validation_ledger_projection_compute/handlers/handler_validation_ledger_projection.py`:
- Around line 121-125: The raise of OnexError when message.value is None (and
the other raises in this file) must be replaced to use the mandatory
error-context factory: create a ModelInfraErrorContext via
ModelInfraErrorContext.with_correlation(transport_type=<appropriate_transport>,
operation="<appropriate_operation>") and pass that context into the OnexError
constructor so the error includes correlation/transport/operation metadata;
update the raise in handler_validation_ledger_projection (the message.value
check) and the other OnexError raises referenced (lines around the other
occurrences) to use the same pattern, ensuring transport_type and operation
values match the handler's transport and operation semantics.
In `@src/omnibase_infra/runtime/db/postgres_validation_ledger_repository.py`:
- Around line 69-81: The ORDER BY in the SQL query constant _SQL_QUERY_BY_RUN_ID
currently sorts only by kafka_partition and kafka_offset, which is
non-deterministic when multiple kafka_topic values exist; update the ORDER BY
clause to include kafka_topic as the first sort key (ORDER BY kafka_topic,
kafka_partition, kafka_offset) and apply the same change to the analogous query
constant used at the other location (the SQL query defined around the second
occurrence referenced in the review), ensuring both constants produce
deterministic replay ordering by (kafka_topic, kafka_partition, kafka_offset).
In `@tests/unit/models/validation_ledger/test_validation_ledger_models.py`:
- Around line 19-27: Add a module-level pytest marker by declaring pytestmark =
pytest.mark.unit at the top of the test module to classify these as unit tests;
ensure the existing import pytest is used (or add it if missing) and place the
pytestmark variable near the other top-level imports so the entire file is
marked with `@pytest.mark.unit`.
In `@tests/unit/nodes/test_node_validation_ledger_projection_compute.py`:
- Around line 24-35: Add a module-level pytest marker so the test file is
classified for CI: declare a top-level variable pytestmark = pytest.mark.unit
near the imports (the file already imports pytest) in
tests/unit/nodes/test_node_validation_ledger_projection_compute.py; this will
mark tests in this module as unit tests—no other code changes to
HandlerValidationLedgerProjection, NodeValidationLedgerProjectionCompute, or
ModelValidationLedgerEntry are needed.
In `@tests/unit/runtime/test_postgres_validation_ledger_repository.py`:
- Around line 22-33: Add a module-level pytest unit marker to this test file by
declaring pytestmark = pytest.mark.unit near the top (after the existing import
pytest) so the whole module is classified as a unit test; locate the file that
imports RepositoryExecutionError, ModelValidationLedgerEntry/Query and
PostgresValidationLedgerRepository and insert the marker there.
🧹 Nitpick comments (4)
src/omnibase_infra/models/validation_ledger/__init__.py (1)
3-11: Minor documentation inconsistency.The docstring lists only 3 models but the package exports 4. Consider updating the docstring to include
ModelValidationLedgerAppendResult.📝 Suggested docstring update
"""Validation event ledger models for cross-repo validation persistence. This package provides domain models for the validation_event_ledger table: - ModelValidationLedgerEntry: Single ledger row - ModelValidationLedgerQuery: Query parameters with optional filters - ModelValidationLedgerReplayBatch: Paginated query result + - ModelValidationLedgerAppendResult: Append operation outcome Ticket: OMN-1908 """src/omnibase_infra/nodes/node_validation_ledger_projection_compute/handlers/handler_validation_ledger_projection.py (2)
182-184: UseOnexErrorwith error context instead ofRuntimeError.Per coding guidelines, use
OnexErrorwithModelInfraErrorContext.with_correlation()for consistent error handling.♻️ Suggested fix
+from omnibase_infra.enums import EnumInfraTransportType +from omnibase_infra.models import ModelInfraErrorContext payload_raw = envelope.get("payload") if not isinstance(payload_raw, dict): - raise RuntimeError("Missing or invalid 'payload' in envelope") + ctx = ModelInfraErrorContext.with_correlation( + transport_type=EnumInfraTransportType.KAFKA, + operation="validation_ledger.project", + correlation_id=correlation_id, + ) + raise OnexError( + "Missing or invalid 'payload' in envelope", + context=ctx, + )As per coding guidelines: "Use MANDATORY error context factory ModelInfraErrorContext.with_correlation() for all exception raises"
269-281: Consider ensuring timezone-awareoccurred_atwhen parsing ISO strings.
datetime.fromisoformat()preserves the input's timezone info. If the input string is timezone-naive (e.g.,"2026-02-07T10:00:00"), the resulting datetime will also be naive, which could cause comparison issues with timezone-aware datetimes elsewhere in the system.🕐 Potential fix to ensure UTC
if isinstance(timestamp_raw, str): - occurred_at = datetime.fromisoformat(timestamp_raw) + parsed = datetime.fromisoformat(timestamp_raw) + # Ensure timezone-aware; assume UTC if naive + if parsed.tzinfo is None: + occurred_at = parsed.replace(tzinfo=UTC) + else: + occurred_at = parsedsrc/omnibase_infra/runtime/db/postgres_validation_ledger_repository.py (1)
173-176: Use strict base64 validation to avoid silent corruption.
base64.b64decodeignores non-base64 characters by default;validate=Truefails fast on malformed envelopes.🛠️ Proposed change
- envelope_bytes_raw = base64.b64decode(entry.envelope_bytes) + envelope_bytes_raw = base64.b64decode(entry.envelope_bytes, validate=True)
Comment on lines
+19
to
+27
| import pytest | ||
| from pydantic import ValidationError | ||
|
|
||
| from omnibase_infra.models.validation_ledger import ( | ||
| ModelValidationLedgerAppendResult, | ||
| ModelValidationLedgerEntry, | ||
| ModelValidationLedgerQuery, | ||
| ModelValidationLedgerReplayBatch, | ||
| ) |
There was a problem hiding this comment.
Add a unit test marker for this module.
These tests are under tests/unit/ but lack an explicit @pytest.mark.unit classification. A module-level marker keeps the file compliant and avoids per-test decoration.
💡 Suggested fix
import pytest
from pydantic import ValidationError
+pytestmark = pytest.mark.unitAs per coding guidelines, "tests/**/*.py: Use pytest markers: @pytest.mark.unit, @pytest.mark.integration, @pytest.mark.slow, @pytest.mark.chaos, @pytest.mark.serial for test classification".
🤖 Prompt for AI Agents
In `@tests/unit/models/validation_ledger/test_validation_ledger_models.py` around
lines 19 - 27, Add a module-level pytest marker by declaring pytestmark =
pytest.mark.unit at the top of the test module to classify these as unit tests;
ensure the existing import pytest is used (or add it if missing) and place the
pytestmark variable near the other top-level imports so the entire file is
marked with `@pytest.mark.unit`.
Comment on lines
+24
to
+35
| import pytest | ||
| import yaml | ||
|
|
||
| from omnibase_core.errors import OnexError | ||
| from omnibase_infra.event_bus.models.model_event_headers import ModelEventHeaders | ||
| from omnibase_infra.event_bus.models.model_event_message import ModelEventMessage | ||
| from omnibase_infra.models.validation_ledger import ModelValidationLedgerEntry | ||
| from omnibase_infra.nodes.node_validation_ledger_projection_compute import ( | ||
| HandlerValidationLedgerProjection, | ||
| NodeValidationLedgerProjectionCompute, | ||
| ) | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
rg -n "pytestmark|@pytest\.mark\." tests/unit/nodes/test_node_validation_ledger_projection_compute.pyRepository: OmniNode-ai/omnibase_infra
Length of output: 52
🏁 Script executed:
cat -n tests/unit/nodes/test_node_validation_ledger_projection_compute.py | head -60Repository: OmniNode-ai/omnibase_infra
Length of output: 2420
🏁 Script executed:
rg -n "^def test_|^class Test" tests/unit/nodes/test_node_validation_ledger_projection_compute.py | head -20Repository: OmniNode-ai/omnibase_infra
Length of output: 245
🏁 Script executed:
sed -n '130,135p;220,225p;422,427p;509,514p;570,575p' tests/unit/nodes/test_node_validation_ledger_projection_compute.pyRepository: OmniNode-ai/omnibase_infra
Length of output: 727
Add module-level pytest marker for test classification.
Missing pytest classification markers causes inconsistent test selection and CI routing. Per coding guidelines, all test files must include pytest markers.
✅ Proposed fix
import pytest
import yaml
+pytestmark = pytest.mark.unit
+
from omnibase_core.errors import OnexError🤖 Prompt for AI Agents
In `@tests/unit/nodes/test_node_validation_ledger_projection_compute.py` around
lines 24 - 35, Add a module-level pytest marker so the test file is classified
for CI: declare a top-level variable pytestmark = pytest.mark.unit near the
imports (the file already imports pytest) in
tests/unit/nodes/test_node_validation_ledger_projection_compute.py; this will
mark tests in this module as unit tests—no other code changes to
HandlerValidationLedgerProjection, NodeValidationLedgerProjectionCompute, or
ModelValidationLedgerEntry are needed.
Take main's reviewed OMN-1908 implementation and align test paths. Remove duplicate test files from branch (model tests, node tests) in favor of main's canonical locations.
- Fixed: 002_drop_validation_event_ledger.sql:14 - index name corrected from idx_validation_ledger_repo_occurred_at to idx_validation_ledger_repo_occurred - Fixed: postgres_validation_ledger_repository.py:306 - docstring updated to reflect kafka_topic, kafka_partition, kafka_offset ordering - Fixed: test_node_ledger_projection_compute.py:344,679,19 - docstrings updated to reference RuntimeHostError instead of OnexError Review iteration: 1/10
…d conventions - Added test_naive_timestamp_gets_utc_timezone for naive-to-UTC conversion path - Propagated correlation_id from headers into _extract_ledger_metadata error contexts - Changed pytest.raises(OnexError) to pytest.raises(RuntimeHostError) for precise assertions - Documented pgcrypto privilege requirement in migration comment - Removed non-canonical NodeValidationLedgerProjectionCompute from registry __all__ Review iteration: 1/10
…error tracing Add UTC timezone to naive ISO timestamps in _extract_timestamp to ensure TIMESTAMPTZ compatibility. Wrap project() body in try/except for error tracing with correlation context. Use validate=True in base64 decoding for stricter input validation.
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/omnibase_infra/nodes/node_ledger_write_effect/handlers/handler_ledger_append.py (1)
307-316:⚠️ Potential issue | 🟠 MajorInclude correlation_id in base64 decode failures
_decode_base64raisesRuntimeHostErrorwithout a correlation_id in the error context, which breaks traceability—especially now thatvalidate=Truewill surface more decode failures. Thread the correlation_id into_decode_base64and include it inModelInfraErrorContext.with_correlation(...).✅ Suggested fix
- def _decode_base64(self, encoded: str) -> bytes: + def _decode_base64(self, encoded: str, *, correlation_id: UUID | None = None) -> bytes: @@ - return base64.b64decode(encoded, validate=True) + return base64.b64decode(encoded, validate=True) except Exception as e: + if correlation_id is None: + correlation_id = uuid4() ctx = ModelInfraErrorContext.with_correlation( + correlation_id=correlation_id, transport_type=EnumInfraTransportType.DATABASE, operation="ledger.append", ) raise RuntimeHostError( f"Failed to decode base64 event data: {type(e).__name__}", context=ctx, ) from e- event_key_bytes = ( - self._decode_base64(payload.event_key) if payload.event_key else None - ) - event_value_bytes = self._decode_base64(payload.event_value) + event_key_bytes = ( + self._decode_base64(payload.event_key, correlation_id=correlation_id) + if payload.event_key + else None + ) + event_value_bytes = self._decode_base64( + payload.event_value, correlation_id=correlation_id + )As per coding guidelines, "Propagate correlation_id from incoming requests; auto-generate with uuid4() if missing; include in all error context".
🤖 Fix all issues with AI agents
In
`@src/omnibase_infra/nodes/node_validation_ledger_projection_compute/handlers/handler_validation_ledger_projection.py`:
- Around line 159-191: The except block should stop embedding the raw exception
string and must propagate an incoming correlation_id into the error context:
capture or derive correlation_id from available sources (e.g.,
self.correlation_id, incoming headers or upstream params) and pass it into
ModelInfraErrorContext.with_correlation(...) (instead of letting it
auto-generate); sanitize any message using the utility in
omnibase_infra.utils.util_error_sanitization (do not include payload-derived
text), and raise RuntimeHostError with a short sanitized message (e.g., include
only exception type and a sanitized summary) and the context built with
EnumInfraTransportType.KAFKA and operation="project_validation_event". Ensure
you still re-raise RuntimeHostError unchanged and only wrap other Exceptions as
described.
…on tracing - Sanitize exception messages via sanitize_error_message() to prevent credential leaks - Add shared correlation_id across error paths in projection handler - Add field_validator for envelope_bytes base64 strictness on ModelValidationLedgerEntry - Fix stale docstring: include kafka_topic in ordering description
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In
`@src/omnibase_infra/nodes/node_validation_ledger_projection_compute/handlers/handler_validation_ledger_projection.py`:
- Around line 194-198: The raised RuntimeHostError in
handler_validation_ledger_projection.py currently prefixes the exception type
and then calls sanitize_error_message(e) which already returns "ExceptionType:
message", causing duplication; change the message construction in the
RuntimeHostError to remove the explicit type prefix and use only
sanitize_error_message(e) (i.e., f"Unexpected error during validation ledger
projection: {sanitize_error_message(e)}"), leaving the context argument and
"from e" chaining intact.
🧹 Nitpick comments (1)
src/omnibase_infra/nodes/node_validation_ledger_projection_compute/handlers/handler_validation_ledger_projection.py (1)
148-163: Consider extracting correlation_id from headers when available.The
_headersparameter is currently unused but could carry an incomingcorrelation_idthat should be propagated per coding guidelines. If upstream Kafka producers include a correlation header, extracting it here would improve distributed tracing continuity.- correlation_id = uuid4() + # Propagate correlation_id from headers if present; auto-generate otherwise + correlation_id: UUID | None = None + if _headers: + raw_cid = _headers.get(b"correlation_id") or _headers.get(b"correlation-id") + if raw_cid: + try: + correlation_id = UUID(raw_cid.decode("utf-8")) + except (ValueError, UnicodeDecodeError): + pass + if correlation_id is None: + correlation_id = uuid4()As per coding guidelines: "Propagate correlation_id from incoming requests; auto-generate with uuid4() if missing; include in all error context"
…idation, explicit error_code
- Fixed: model_validation_ledger_entry.py:61 - Strip RFC 2045 newlines before strict
base64 validation so PostgreSQL-produced base64 passes the check
- Fixed: model_validation_ledger_entry.py:47 - Enforce SHA-256 hex format (^[0-9a-f]{64}$)
on envelope_hash field
- Fixed: handler_validation_ledger_projection.py:194 - Explicit error_code=OPERATION_FAILED
on catch-all exception handler
- Updated test fixtures to use valid SHA-256 hash constants
Review iteration: 1/10
- Fixed: postgres_validation_ledger_repository.py:172 - Docstring example
showed "sha256:abc123" which doesn't match the ^[0-9a-f]{64}$ pattern
now enforced by ModelValidationLedgerEntry
Review iteration: 2/10
…mples - Remove redundant type(e).__name__ prefix from handler error message (original exception type preserved via __cause__ chain) - Fix protocol docstring example: envelope_hash must be raw SHA-256 hex digest, not "sha256:abc123" format - Fix test helper _append_kwargs to use valid SHA-256 hex digest - Remove stale .pyc for non-existent node test file
…sistency - Extract correlation_id from Kafka headers for distributed tracing instead of always generating a new uuid4 - Remove redundant exception type prefix from sanitized error messages - Expand model docstrings (append result, query, replay batch, entry) - Fix envelope_hash field description to clarify pre-encoding hash - Update protocol docstring to reflect kafka_topic in ORDER BY - Add SPDX header to drop migration, update module __init__ docs
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/omnibase_infra/protocols/protocol_validation_ledger_repository.py (1)
15-15:⚠️ Potential issue | 🟡 MinorModule docstring ordering description is inconsistent with method docstrings.
Line 15 states "Deterministic replay: Ordering by kafka_partition, kafka_offset" but the updated method docstrings (lines 118-119, 128-129) correctly describe ordering as "kafka_topic, kafka_partition, kafka_offset". The module-level docstring should be updated for consistency.
📝 Proposed fix
- - Deterministic replay: Ordering by kafka_partition, kafka_offset + - Deterministic replay: Ordering by kafka_topic, kafka_partition, kafka_offset
jonahgabriel
deleted the
jonah/omn-1908-cross-repo-validation-ledger-persistence
branch
February 7, 2026 19:12
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
Components Added
migrations/002_create_validation_event_ledger.sqlmigrations/002_drop_validation_event_ledger.sqlmodels/validation_ledger/protocols/protocol_validation_ledger_repository.pyruntime/db/postgres_validation_ledger_repository.pynodes/node_validation_ledger_projection_compute/tests/unit/Test plan
pytest -xvs)repo_id: str)Summary by CodeRabbit
Chores
Bug Fixes
Tests