Repository navigation
feat(projection): track last_heartbeat_at in registration projection [OMN-1006] - #94
Conversation
…[OMN-1006] Add last_heartbeat_at field to track when the last heartbeat was received from a node, enabling accurate reporting in NodeLivenessExpired events. Changes: - Add last_heartbeat_at field to ModelRegistrationProjection - Add last_heartbeat_at TIMESTAMPTZ column to SQL schema - Update projector upsert to include last_heartbeat_at - Add update_heartbeat() method to projector for heartbeat processing - Update projection reader to populate last_heartbeat_at from DB rows - Update timeout emitter to use projection.last_heartbeat_at - Add heartbeat handler (draft, wiring in follow-up) - Fix test fixtures to include new field - Update INFRA_MAX_UNIONS threshold (589 -> 600) Acceptance Criteria: - [x] ModelRegistrationProjection has last_heartbeat_at field - [x] SQL schema includes last_heartbeat_at column - [x] Projector supports updating last_heartbeat_at - [x] ModelNodeLivenessExpired includes accurate last_heartbeat_at - [x] All existing tests pass
|
Warning Rate limit exceeded@jonahgabriel has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 7 minutes and 35 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📒 Files selected for processing (9)
📝 WalkthroughWalkthroughAdds node heartbeat support: new projection field Changes
Sequence DiagramsequenceDiagram
autonumber
participant Event as Heartbeat Event
participant Handler as HandlerNodeHeartbeat
participant Reader as ProjectionReader
participant Projector as ProjectorRegistration
participant DB as Database
Note over Handler,Projector: New heartbeat flow (OMN-1006)
Event->>Handler: receive ModelNodeHeartbeatEvent
activate Handler
Handler->>Reader: fetch projection by node_id
activate Reader
Reader->>DB: SELECT registration_projections
DB-->>Reader: row (includes last_heartbeat_at, liveness_deadline, state)
Reader-->>Handler: ModelRegistrationProjection
deactivate Reader
alt projection not found
Handler-->>Event: ModelHeartbeatHandlerResult(node_not_found=true)
else projection found
Handler->>Handler: compute new_liveness_deadline = event.timestamp + window
Handler->>Projector: update_heartbeat(node_id, domain, last_heartbeat_at, liveness_deadline, correlation_id)
activate Projector
Projector->>DB: UPDATE registration_projections SET last_heartbeat_at, liveness_deadline ...
DB-->>Projector: rows affected
Projector-->>Handler: bool (updated?)
deactivate Projector
alt updated
Handler-->>Event: ModelHeartbeatHandlerResult(success=true, previous_state, last_heartbeat_at, liveness_deadline)
else not updated
Handler-->>Event: ModelHeartbeatHandlerResult(success=false, node_not_found=true)
end
end
deactivate Handler
Estimated code review effort🎯 4 (Complex) | ⏱️ ~40 minutes Poem
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
🧹 Nitpick comments (1)
src/omnibase_infra/orchestrators/registration/handlers/__init__.py (1)
5-13: Consider re-exportingDEFAULT_LIVENESS_WINDOW_SECONDSfor API consistency.The
handler_node_heartbeat.pymodule includesDEFAULT_LIVENESS_WINDOW_SECONDSin its__all__(line 299), but the package__init__.pydoes not re-export it. This creates an inconsistency where users can accessfrom omnibase_infra.orchestrators.registration.handlers.handler_node_heartbeat import DEFAULT_LIVENESS_WINDOW_SECONDSbut not from the package level.Either add the constant to this file's exports, or remove it from the module's
__all__if it's meant to be internal.🔎 Suggested fix to include the constant
from omnibase_infra.orchestrators.registration.handlers.handler_node_heartbeat import ( + DEFAULT_LIVENESS_WINDOW_SECONDS, HandlerNodeHeartbeat, ModelHeartbeatHandlerResult, ) __all__ = [ + "DEFAULT_LIVENESS_WINDOW_SECONDS", "HandlerNodeHeartbeat", "ModelHeartbeatHandlerResult", ]
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Lite
📒 Files selected for processing (12)
src/omnibase_infra/models/projection/model_registration_projection.pysrc/omnibase_infra/orchestrators/__init__.pysrc/omnibase_infra/orchestrators/registration/__init__.pysrc/omnibase_infra/orchestrators/registration/handlers/__init__.pysrc/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.pysrc/omnibase_infra/projectors/projection_reader_registration.pysrc/omnibase_infra/projectors/projector_registration.pysrc/omnibase_infra/schemas/schema_registration_projection.sqlsrc/omnibase_infra/services/timeout_emitter.pysrc/omnibase_infra/validation/infra_validators.pytests/unit/projectors/test_projection_reader_registration.pytests/unit/validation/test_validator_defaults.py
🧰 Additional context used
📓 Path-based instructions (1)
**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.py: NEVER useAnytypes in Python - Always use specific types, useobjectfor generic dispatchers instead
All data structures must be proper Pydantic models - one model per file
Use nullable type annotationX | None(PEP 604 union syntax) overOptional[X]for null types in Python
All services MUST useModelONEXContainerfor container-based dependency injection
Error classes must raiseOnexErrornot base Exception - useraise OnexError(...) from epattern
Never useisinstancefor protocol resolution - use duck typing through protocols instead
NEVER include passwords, API keys, tokens, secrets, full connection strings with credentials, PII, or private keys in error messages or context
Files:
src/omnibase_infra/orchestrators/__init__.pysrc/omnibase_infra/orchestrators/registration/handlers/__init__.pysrc/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.pysrc/omnibase_infra/projectors/projector_registration.pysrc/omnibase_infra/orchestrators/registration/__init__.pytests/unit/projectors/test_projection_reader_registration.pysrc/omnibase_infra/models/projection/model_registration_projection.pysrc/omnibase_infra/projectors/projection_reader_registration.pysrc/omnibase_infra/validation/infra_validators.pytests/unit/validation/test_validator_defaults.pysrc/omnibase_infra/services/timeout_emitter.py
🧠 Learnings (5)
📚 Learning: 2025-12-03T16:55:49.755Z
Learnt from: CR
Repo: OmniNode-ai/omniclaude PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-03T16:55:49.755Z
Learning: Applies to agents/**/*.py : Implement ONEX-compliant agent architecture with four node types: Effect (External I/O), Compute (Pure transforms), Reducer (State/persistence), and Orchestrator (Workflow coordination)
Applied to files:
src/omnibase_infra/orchestrators/__init__.py
📚 Learning: 2025-11-24T16:32:55.606Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T16:32:55.606Z
Learning: ORCHESTRATOR Nodes must inherit from `NodeOrchestrator` or use `NodeOrchestratorService` and must coordinate workflows, manage node interactions, and handle process/event orchestration
Applied to files:
src/omnibase_infra/orchestrators/__init__.py
📚 Learning: 2025-11-24T17:23:49.777Z
Learnt from: CR
Repo: OmniNode-ai/omniclaude PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T17:23:49.777Z
Learning: All ONEX nodes must conform to the canonical structure, code generation, and interface patterns established in the `node_cli` node, using it as the primary source of truth for directory structure, contract schema patterns, linked document architecture, base state patterns, shared schema references, extensibility patterns, CLI interface declarations, code generation, dependency injection, error handling, testing, and documentation
Applied to files:
src/omnibase_infra/orchestrators/__init__.py
📚 Learning: 2025-12-07T17:50:13.678Z
Learnt from: CR
Repo: OmniNode-ai/omniintelligence PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-07T17:50:13.678Z
Learning: Decompose intelligence operations into specialized ONEX nodes following a four-node pattern: Orchestrator (coordinate workflows), Reducer (manage state, FSM transitions), Compute (pure data processing), and Effect (external I/O)
Applied to files:
src/omnibase_infra/orchestrators/__init__.py
📚 Learning: 2025-12-06T22:21:32.649Z
Learnt from: CR
Repo: OmniNode-ai/omniagent PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-06T22:21:32.649Z
Learning: Applies to nodes/**/*.py : Use `omnibase_infra` handlers for OmniIntelligence queries via HttpRestAdapter envelope pattern
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
🧬 Code graph analysis (3)
src/omnibase_infra/orchestrators/registration/handlers/__init__.py (1)
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py (2)
HandlerNodeHeartbeat(98-293)ModelHeartbeatHandlerResult(45-95)
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py (6)
src/omnibase_infra/enums/enum_infra_transport_type.py (1)
EnumInfraTransportType(28-52)src/omnibase_infra/enums/enum_registration_state.py (1)
EnumRegistrationState(24-258)src/omnibase_infra/errors/infra_errors.py (1)
RuntimeHostError(35-100)src/omnibase_infra/models/registration/model_node_heartbeat_event.py (1)
ModelNodeHeartbeatEvent(19-95)src/omnibase_infra/projectors/projection_reader_registration.py (2)
ProjectionReaderRegistration(45-656)get_entity_state(146-226)src/omnibase_infra/projectors/projector_registration.py (2)
ProjectorRegistration(52-840)update_heartbeat(717-840)
src/omnibase_infra/orchestrators/registration/__init__.py (1)
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py (2)
HandlerNodeHeartbeat(98-293)ModelHeartbeatHandlerResult(45-95)
🔇 Additional comments (17)
src/omnibase_infra/models/projection/model_registration_projection.py (1)
137-140: LGTM! Field addition is clean and well-documented.The
last_heartbeat_atfield is properly defined with:
- Correct type annotation using PEP 604 union syntax (
datetime | None)- Appropriate default value (
None)- Clear description for liveness reporting
- Logical placement between deadline and emission marker fields
src/omnibase_infra/projectors/projection_reader_registration.py (1)
134-134: LGTM! Database row mapping is correct.The
last_heartbeat_atfield is properly extracted from the database row and passed to the projection constructor, following the same pattern as other timestamp fields.src/omnibase_infra/validation/infra_validators.py (1)
373-377: LGTM! Threshold adjustment accounts for new heartbeat functionality.The
INFRA_MAX_UNIONSthreshold is updated from 589 to 600 to accommodate the new heartbeat handler and projection update logic. The 11-unit increase (for ~6 unions mentioned) provides a reasonable buffer above the current baseline for near-term codebase growth, which aligns with the stated strategy.src/omnibase_infra/services/timeout_emitter.py (1)
626-626: LGTM! Liveness event now includes actual heartbeat timestamp.The event now correctly uses
projection.last_heartbeat_atinstead ofNone, providing accurate reporting of when the last heartbeat was received. The field is properly nullable (datetime | None), so the event can handle cases where no heartbeat was ever received.tests/unit/projectors/test_projection_reader_registration.py (1)
88-88: LGTM! Test mock data updated correctly.The
last_heartbeat_atfield is properly added to the mock row with aNonedefault, maintaining compatibility with the updated projection schema while allowing tests to override when needed.src/omnibase_infra/orchestrators/__init__.py (1)
1-3: LGTM! Package initializer follows conventions.The orchestrator package
__init__.pyis properly structured with license headers and a clear docstring describing its purpose.src/omnibase_infra/schemas/schema_registration_projection.sql (2)
67-67: LGTM! Schema column added correctly.The
last_heartbeat_atcolumn is properly defined asTIMESTAMPTZ(timezone-aware) and nullable, matching the model definition. The placement betweenliveness_deadlineand the timeout emission markers is logical.Note: Ensure that any database migration strategy accounts for adding this column to existing
registration_projectionstables. The schema is idempotent for new installations, but existing deployments may need explicit migration steps.
175-177: LGTM! Column documentation is clear.The comment accurately describes the purpose of
last_heartbeat_atand when it's updated, providing helpful context for database administrators and future developers.src/omnibase_infra/orchestrators/registration/__init__.py (1)
1-13: LGTM! Registration orchestrator package exports are clean.The registration orchestrator package properly re-exports
HandlerNodeHeartbeatandModelHeartbeatHandlerResult, making them available for integration. The structure follows Python conventions with clear__all__definition.Note: The PR description mentions the heartbeat handler is "draft" and "not yet wired into event routing." This export structure supports future integration when the handler is ready to be activated.
tests/unit/validation/test_validator_defaults.py (1)
58-65: LGTM!The threshold adjustment is well-documented with ticket reference (OMN-1006), follows the established history pattern, and the assertion is consistent with the updated value.
src/omnibase_infra/projectors/projector_registration.py (2)
259-278: LGTM!The upsert SQL correctly incorporates
last_heartbeat_at:
- Added to INSERT column list (line 262)
- Placeholder updated with correct position
$9(line 268)- ON CONFLICT SET clause properly references
EXCLUDED.last_heartbeat_at(line 278)
717-840: LGTM!The new
update_heartbeatmethod is well-implemented:
- Follows the established pattern from
update_ack_timeout_markerandupdate_liveness_timeout_marker- Atomically updates both
last_heartbeat_atandliveness_deadline- Proper circuit breaker integration
- Consistent error handling with typed exceptions
- Comprehensive docstring with example and ticket references
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py (5)
1-42: LGTM!Module structure is well-organized with proper license header, docstring with ticket references, and clean imports. The
TYPE_CHECKINGusage correctly avoids circular import issues with projector/reader dependencies.
45-95: LGTM!The result model follows project conventions:
- Immutable with
frozen=Trueandextra="forbid"- Uses PEP 604 union syntax (
X | None) per coding guidelines- All fields have descriptive Field definitions
133-155: LGTM!Clean initialization with sensible default for
liveness_window_seconds. The 90-second default (3x the 30-second heartbeat interval) is well-documented and allows for 2 missed heartbeats before liveness expiry.
193-280: LGTM!The happy path implementation is well-structured:
- Gracefully handles missing projections with descriptive error messages
- Appropriately logs warnings for non-active nodes while still processing (handles race conditions)
- Correctly calculates new liveness deadline from event timestamp
- Returns comprehensive result with previous state and updated timestamps
296-300: LGTM!Exports are appropriate for the module's public API.
Merge: - Integrate OMN-816 handler naming changes from main PR Review Fixes: - Fix exception handling to preserve InfraConnectionError/InfraTimeoutError types - Re-export DEFAULT_LIVENESS_WINDOW_SECONDS from handlers __init__.py
PR Review: feat(projection): track last_heartbeat_at in registration projection [OMN-1006]SummaryThis PR adds heartbeat tracking to the registration projection system, enabling accurate reporting of when nodes last sent heartbeats in liveness expiration events. The implementation is well-architected and follows ONEX infrastructure patterns closely. ✅ Strengths1. Excellent ONEX Compliance
2. Clean Architecture
3. Database Design
4. Documentation Quality
🔍 Code Quality Observationshandler_node_heartbeat.py:284-289Issue: Exception handling preserves The PR description mentions "Fix exception handling to preserve InfraConnectionError/InfraTimeoutError types" - this is correctly implemented: except InfraConnectionError:
# Re-raise infrastructure connection errors directly (preserve error type)
raise
except InfraTimeoutError:
# Re-raise infrastructure timeout errors directly (preserve error type)
raiseThis follows ONEX error handling patterns perfectly. projector_registration.py:717-840Strong implementation of
timeout_emitter.py:626Excellent fix - Now passes last_heartbeat_at=projection.last_heartbeat_at, # Was: NoneThis directly fulfills the ticket's acceptance criteria. 🎯 Performance Considerations✅ Optimized Operations
💡 Potential Optimization (Future)The handler performs a read-then-write pattern: projection = await self._projection_reader.get_entity_state(...) # SELECT
# ...
updated = await self._projector.update_heartbeat(...) # UPDATEConsideration: For high-throughput heartbeat processing, this could be optimized to a single UPDATE with
Recommendation: Keep current implementation unless profiling shows heartbeat processing is a bottleneck. 🔒 Security Review✅ No Security Concerns
🧪 Test Coverage✅ Test Updates
|
- Fix exception handling to preserve all RuntimeHostError subtypes (InfraConnectionError, InfraTimeoutError, etc.) instead of wrapping - Wire HandlerNodeHeartbeat into registration orchestrator with set_heartbeat_handler(), has_heartbeat_handler, handle_heartbeat() - Add direct_handler flag to contract.yaml for non-workflow events - Add 27 integration tests for heartbeat handler (8 test classes) - Verify ModelNodeLivenessExpired timestamps are accurate (documented) - Re-export DEFAULT_LIVENESS_WINDOW_SECONDS for API consistency - Update orchestrator tests to whitelist heartbeat handler methods
PR Review: Track last_heartbeat_at in Registration Projection [OMN-1006]SummaryThis PR adds heartbeat timestamp tracking to the registration projection, enabling accurate liveness reporting. The implementation follows ONEX infrastructure patterns and includes comprehensive test coverage (877 lines of integration tests). ✅ Strengths1. Excellent ONEX Compliance
2. Database Design
3. Error Handling Excellenceexcept RuntimeHostError:
# Re-raise all infrastructure errors directly (preserves error type)
raise
except Exception as e:
# Wrap only non-infrastructure errors
raise RuntimeHostError(...) from e
4. Test Coverage
5. Documentation Quality
🔍 Issues & Recommendations1. 🟡 Medium: Handler Wiring Not DemonstratedIssue: The heartbeat handler is created but wiring to event routing is deferred to "follow-up". This creates integration risk. Location: Recommendation: # Consider adding integration test showing full event flow:
# 1. Heartbeat event received from Kafka
# 2. Dispatcher routes to orchestrator.handle_heartbeat()
# 3. Projection updated
# 4. Liveness deadline extendedRisk: Medium - functionality exists but integration path unverified 2. 🟡 Medium: Liveness Window Calculation InconsistencyIssue: Deadline calculated from Location: new_liveness_deadline = heartbeat_timestamp + timedelta(
seconds=self._liveness_window_seconds
)Problem: If event processing is delayed (e.g., Kafka lag), deadline may already be expired when calculated. Example:
Recommendation: # Option 1: Use processing time (safer for delayed events)
now = datetime.now(UTC)
new_liveness_deadline = now + timedelta(seconds=self._liveness_window_seconds)
# Option 2: Use max(event_time, now) to handle both cases
reference_time = max(heartbeat_timestamp, datetime.now(UTC))
new_liveness_deadline = reference_time + timedelta(seconds=self._liveness_window_seconds)Impact: Could cause false liveness expirations in high-latency scenarios 3. 🟢 Low: Unused Variable in Error ContextIssue: Location: Recommendation: Consider lazy construction or clarify intent with comment: # Error context prepared for potential exception handling
ctx = ModelInfraErrorContext(...)4. 🟢 Low: Test Helper Function NamingLocation: def make_projection(ONEX Convention: Factory functions should be Recommendation: Consider renaming to 5. 🟢 Low: Union Validator Threshold BumpLocation: INFRA_MAX_UNIONS = 600 # Was 589Question: Is this increase due to new heartbeat models, or unrelated growth? Recommendation: Add comment explaining what pushed the limit: # OMN-1006: Heartbeat handler models increased union count by ~11
INFRA_MAX_UNIONS = 600🔒 Security Review✅ No security concerns identified
🚀 Performance ConsiderationsPositive:
Watch:
📊 Code Quality Metrics
🎯 Acceptance Criteria ReviewFrom PR description:
🔧 Recommendations SummaryMust Address Before Merge:
Should Address (Minor):
Nice to Have:
✅ Final VerdictAPPROVE with minor recommendations This is high-quality ONEX infrastructure code with excellent test coverage and documentation. The liveness window calculation issue (#2) should be addressed before merge, but it's a design decision rather than a bug. Once resolved, this PR is production-ready. Suggested Next Steps:
Great work on the comprehensive test suite and adherence to ONEX patterns! 🚀 Reviewed against: CLAUDE.md ONEX Infrastructure Guidelines |
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
tests/integration/registration/handlers/conftest.py (1)
69-94: Prefer importing DEFAULT_LIVENESS_WINDOW_SECONDS constant.Line 93 hardcodes
90.0for the liveness window. For consistency and maintainability, import and useDEFAULT_LIVENESS_WINDOW_SECONDSfrom the handlers package instead.🔎 Proposed refactor
@pytest.fixture def heartbeat_handler( reader: ProjectionReaderRegistration, projector: ProjectorRegistration, ) -> HandlerNodeHeartbeat: """Function-scoped HandlerNodeHeartbeat instance. Creates a handler with the default liveness window (90 seconds). Suitable for most integration tests. Args: reader: ProjectionReaderRegistration fixture for state lookups. projector: ProjectorRegistration fixture for state updates. Returns: HandlerNodeHeartbeat configured with default liveness window. """ from omnibase_infra.orchestrators.registration.handlers import ( + DEFAULT_LIVENESS_WINDOW_SECONDS, HandlerNodeHeartbeat, ) return HandlerNodeHeartbeat( projection_reader=reader, projector=projector, - liveness_window_seconds=90.0, + liveness_window_seconds=DEFAULT_LIVENESS_WINDOW_SECONDS, )tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py (2)
555-589: Consider adding a more specific assertion for rapid heartbeat test.The test correctly handles non-deterministic ordering with the comment at lines 587-588. However, you could strengthen the test by verifying that the final
last_heartbeat_atis within the expected range of event timestamps.🔎 Optional: Add timestamp range verification
# Final state should have the last heartbeat timestamp final = await reader.get_entity_state(node_id) assert final is not None assert final.last_heartbeat_at is not None # The last heartbeat timestamp should be one of the event timestamps # (exact order is non-deterministic with concurrent writes) + # Verify the timestamp is within the expected range + earliest_time = base_time + latest_time = base_time + timedelta(milliseconds=900) + assert earliest_time <= final.last_heartbeat_at <= latest_time
729-731: Consider catching a more specific exception type.The test catches a generic
Exceptionwhen verifying the model is frozen. Pydantic frozen models raisepydantic.ValidationErrorwhen attempting to modify them.🔎 Optional: Use specific exception type
+from pydantic import ValidationError + # Attempt to modify should fail - with pytest.raises(Exception): # ValidationError for frozen models + with pytest.raises(ValidationError): result.success = False # type: ignore[misc]
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Lite
📒 Files selected for processing (17)
src/omnibase_infra/nodes/node_registration_orchestrator/__init__.pysrc/omnibase_infra/nodes/node_registration_orchestrator/contract.yamlsrc/omnibase_infra/nodes/node_registration_orchestrator/models/model_node_liveness_expired.pysrc/omnibase_infra/nodes/node_registration_orchestrator/node.pysrc/omnibase_infra/orchestrators/registration/__init__.pysrc/omnibase_infra/orchestrators/registration/handlers/__init__.pysrc/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.pysrc/omnibase_infra/projectors/projection_reader_registration.pysrc/omnibase_infra/projectors/projector_registration.pysrc/omnibase_infra/validation/infra_validators.pytests/integration/registration/handlers/__init__.pytests/integration/registration/handlers/conftest.pytests/integration/registration/handlers/test_handler_node_heartbeat_integration.pytests/unit/nodes/test_node_registration_orchestrator.pytests/unit/nodes/test_orchestrator_decision_paths.pytests/unit/nodes/test_orchestrator_no_io.pytests/unit/validation/test_validator_defaults.py
✅ Files skipped from review due to trivial changes (2)
- src/omnibase_infra/nodes/node_registration_orchestrator/models/model_node_liveness_expired.py
- tests/integration/registration/handlers/init.py
🚧 Files skipped from review as they are similar to previous changes (3)
- src/omnibase_infra/projectors/projection_reader_registration.py
- src/omnibase_infra/validation/infra_validators.py
- src/omnibase_infra/orchestrators/registration/init.py
🧰 Additional context used
📓 Path-based instructions (1)
**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.py: NEVER useAnytypes in Python - Always use specific types, useobjectfor generic dispatchers instead
All data structures must be proper Pydantic models - one model per file
Use nullable type annotationX | None(PEP 604 union syntax) overOptional[X]for null types in Python
All services MUST useModelONEXContainerfor container-based dependency injection
Error classes must raiseOnexErrornot base Exception - useraise OnexError(...) from epattern
Never useisinstancefor protocol resolution - use duck typing through protocols instead
NEVER include passwords, API keys, tokens, secrets, full connection strings with credentials, PII, or private keys in error messages or context
Files:
tests/integration/registration/handlers/conftest.pysrc/omnibase_infra/projectors/projector_registration.pytests/integration/registration/handlers/test_handler_node_heartbeat_integration.pysrc/omnibase_infra/orchestrators/registration/handlers/__init__.pysrc/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.pytests/unit/validation/test_validator_defaults.pytests/unit/nodes/test_orchestrator_no_io.pytests/unit/nodes/test_orchestrator_decision_paths.pytests/unit/nodes/test_node_registration_orchestrator.pysrc/omnibase_infra/nodes/node_registration_orchestrator/node.pysrc/omnibase_infra/nodes/node_registration_orchestrator/__init__.py
🧠 Learnings (26)
📚 Learning: 2025-12-07T17:50:13.678Z
Learnt from: CR
Repo: OmniNode-ai/omniintelligence PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-07T17:50:13.678Z
Learning: Organize tests following the structure: tests/conftest.py for shared fixtures, tests/unit/ for unit tests (no infrastructure), tests/integration/ for integration tests (requires Kafka/DBs), tests/nodes/ for node-specific tests
Applied to files:
tests/integration/registration/handlers/conftest.py
📚 Learning: 2025-11-30T21:55:10.298Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-11-30T21:55:10.298Z
Learning: Applies to src/omninode_bridge/nodes/**/contract.yaml : All contract YAML files for ONEX v2.0 nodes MUST define subcontract references, input/output models, and FSM configurations. Use YAML 1.2 syntax.
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/contract.yaml
📚 Learning: 2025-11-24T17:23:49.777Z
Learnt from: CR
Repo: OmniNode-ai/omniclaude PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T17:23:49.777Z
Learning: Applies to **/node_*/v[0-9]*_[0-9]*_[0-9]*/contracts/contract_*.yaml : All ONEX node contract definitions must use the new subcontract architecture pattern, breaking down complex contracts into separate contract_actions.yaml, contract_models.yaml, contract_validation.yaml, and optional contract_cli.yaml and contract_capabilities.yaml files for separation of concerns, maintainability, reusability, modularity, and future tool-as-a-service readiness
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/contract.yaml
📚 Learning: 2025-11-24T17:23:49.777Z
Learnt from: CR
Repo: OmniNode-ai/omniclaude PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T17:23:49.777Z
Learning: Applies to **/node_*/v[0-9]*_[0-9]*_[0-9]*/contract.yaml : All ONEX node contract definitions must follow the linked document architecture pattern with contract.yaml linking to node_config.yaml and deployment_config.yaml as associated documents
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/contract.yaml
📚 Learning: 2025-11-24T17:23:49.777Z
Learnt from: CR
Repo: OmniNode-ai/omniclaude PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T17:23:49.777Z
Learning: Applies to **/node_*/v[0-9]*_[0-9]*_[0-9]*/contracts/contract_capabilities.yaml : All ONEX node execution capability definitions, if applicable, must be included in contract_capabilities.yaml with supported_node_types, supported_delivery_modes, and performance_constraints specifications
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/contract.yaml
📚 Learning: 2025-11-30T21:55:10.298Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-11-30T21:55:10.298Z
Learning: Applies to src/omninode_bridge/events/**/*.py : Kafka event publishing MUST use OnexEnvelopeV1 format with 13 topics for event streaming at all workflow lifecycle stages
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/contract.yaml
📚 Learning: 2025-11-24T16:32:55.606Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T16:32:55.606Z
Learning: Applies to tests/unit/infrastructure/**/test_*.py : All node implementations must have comprehensive unit tests following the testing pattern in `tests/unit/infrastructure/` with tests for node initialization and node execution
Applied to files:
tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py
📚 Learning: 2025-12-25T19:10:27.034Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.034Z
Learning: Applies to **/*adapter*.py **/*handler*.py : Use `InfraConnectionError` for connection failures, with transport-aware error codes (DATABASE_CONNECTION_ERROR for database, NETWORK_ERROR for HTTP/GRPC, SERVICE_UNAVAILABLE for Kafka/Consul/Vault/Valkey)
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
📚 Learning: 2025-12-25T19:10:27.034Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.034Z
Learning: Applies to **/*adapter*.py **/*handler*.py : Use `InfraTimeoutError` for operation timeouts
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
📚 Learning: 2025-12-25T19:10:27.034Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.034Z
Learning: Applies to **/*adapter*.py **/*handler*.py : Use graceful degradation for `InfraTimeoutError` - fallback to secondary data source (cache, secondary database) when primary times out
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
📚 Learning: 2025-12-25T19:10:27.034Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.034Z
Learning: Applies to **/*adapter*.py **/*handler*.py : Use `InfraUnavailableError` for service unavailable conditions, including when circuit breaker is open
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
📚 Learning: 2025-12-25T19:10:27.033Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.033Z
Learning: Applies to **/*adapter*.py **/*handler*.py **/*service*.py : Use `ModelInfraErrorContext` with `transport_type` when raising infrastructure errors
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
📚 Learning: 2025-12-25T19:10:27.034Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.034Z
Learning: Applies to **/*adapter*.py **/*handler*.py : Use `InfraAuthenticationError` for authentication/authorization failures
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
📚 Learning: 2025-12-25T19:10:27.034Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.034Z
Learning: Applies to **/*adapter*.py **/*handler*.py : Use circuit breaker pattern for `InfraUnavailableError` to prevent cascading failures - prevent requests when circuit is open, give service time to recover
Applied to files:
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py
📚 Learning: 2025-11-30T21:55:10.298Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-11-30T21:55:10.298Z
Learning: Applies to src/omninode_bridge/codegen/**/*.py : Code generation service MUST auto-generate ONEX v2.0 compliant nodes with intelligent mixin injection and quality validation. Generate comprehensive test suites with 90%+ coverage.
Applied to files:
tests/unit/nodes/test_node_registration_orchestrator.py
📚 Learning: 2025-11-24T16:32:55.606Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T16:32:55.606Z
Learning: ORCHESTRATOR Nodes must inherit from `NodeOrchestrator` or use `NodeOrchestratorService` and must coordinate workflows, manage node interactions, and handle process/event orchestration
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.pysrc/omnibase_infra/nodes/node_registration_orchestrator/__init__.py
📚 Learning: 2025-12-20T04:09:41.832Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_core PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-20T04:09:41.832Z
Learning: Applies to **/*.py : Use ModelONEXContainer in node constructors for dependency injection, never use ModelContainer[T]
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-11-30T21:55:10.298Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-11-30T21:55:10.298Z
Learning: Applies to src/omninode_bridge/nodes/**/*.py : All nodes in omninode_bridge MUST use omnibase_core standards (ModelServiceEffect, ModelServiceCompute for effect/compute nodes; NodeOrchestrator, NodeReducer with mixins for orchestrator/reducer nodes)
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-11-30T21:55:10.298Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-11-30T21:55:10.298Z
Learning: Applies to src/omninode_bridge/nodes/**/*.py : Import mixins from omnibase_core.mixins.* and use Mixin* naming pattern (e.g., MixinHealthCheck, MixinMetrics, MixinEventBus) - never use local custom mixins unless experimental and documented
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-11-24T16:32:55.606Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T16:32:55.606Z
Learning: Node communication must use event-driven patterns through `ModelEventEnvelope` from `omnibase_core.models.events.model_event_envelope`
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-12-25T19:10:27.034Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.034Z
Learning: Applies to nodes/**/node.py : Node archetypes and I/O models must be imported from `omnibase_core.nodes`, never defined in infra - NodeEffect, NodeCompute, NodeReducer, NodeOrchestrator
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-12-08T00:48:30.737Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_spi PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-08T00:48:30.737Z
Learning: Import `omnibase_core` models and types only for type hints and runtime usage - follow the SPI → Core dependency direction
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-12-25T19:10:27.033Z
Learnt from: CR
Repo: OmniNode-ai/omnibase_infra PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-25T19:10:27.033Z
Learning: Applies to **/*.py : All services MUST use `ModelONEXContainer` for container-based dependency injection
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-11-24T16:32:55.606Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T16:32:55.606Z
Learning: Node services must implement dependency injection using `ModelONEXContainer` from `omnibase_core.models.container.model_onex_container`
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-12-06T22:21:32.649Z
Learnt from: CR
Repo: OmniNode-ai/omniagent PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-06T22:21:32.649Z
Learning: Applies to nodes/**/*.py : Use event bus mixins from `omnibase_core` for Kafka publishing instead of direct Kafka clients
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.py
📚 Learning: 2025-12-03T16:55:49.755Z
Learnt from: CR
Repo: OmniNode-ai/omniclaude PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-03T16:55:49.755Z
Learning: Applies to agents/**/*.py : Implement ONEX-compliant agent architecture with four node types: Effect (External I/O), Compute (Pure transforms), Reducer (State/persistence), and Orchestrator (Workflow coordination)
Applied to files:
src/omnibase_infra/nodes/node_registration_orchestrator/node.pysrc/omnibase_infra/nodes/node_registration_orchestrator/__init__.py
🧬 Code graph analysis (4)
src/omnibase_infra/projectors/projector_registration.py (4)
src/omnibase_infra/errors/model_infra_error_context.py (1)
ModelInfraErrorContext(17-96)src/omnibase_infra/enums/enum_infra_transport_type.py (1)
EnumInfraTransportType(28-52)src/omnibase_infra/mixins/mixin_async_circuit_breaker.py (3)
_check_circuit_breaker(271-366)_reset_circuit_breaker(458-533)_record_circuit_failure(368-456)src/omnibase_infra/errors/infra_errors.py (3)
InfraConnectionError(181-286)InfraTimeoutError(289-326)RuntimeHostError(35-100)
tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py (10)
src/omnibase_infra/errors/infra_errors.py (2)
InfraConnectionError(181-286)RuntimeHostError(35-100)src/omnibase_infra/models/projection/model_registration_projection.py (1)
ModelRegistrationProjection(35-332)src/omnibase_infra/models/projection/model_sequence_info.py (1)
ModelSequenceInfo(21-179)src/omnibase_infra/models/registration/model_node_heartbeat_event.py (1)
ModelNodeHeartbeatEvent(19-95)src/omnibase_infra/models/registration/model_node_capabilities.py (1)
ModelNodeCapabilities(13-167)src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py (3)
HandlerNodeHeartbeat(98-305)ModelHeartbeatHandlerResult(45-95)handle(157-305)src/omnibase_infra/projectors/projection_reader_registration.py (1)
ProjectionReaderRegistration(45-656)tests/integration/registration/handlers/conftest.py (2)
heartbeat_handler(70-94)heartbeat_handler_fast_window(98-122)src/omnibase_infra/enums/enum_infra_transport_type.py (1)
EnumInfraTransportType(28-52)tests/integration/projectors/conftest.py (1)
pg_pool(142-185)
src/omnibase_infra/orchestrators/registration/handlers/__init__.py (1)
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py (2)
HandlerNodeHeartbeat(98-305)ModelHeartbeatHandlerResult(45-95)
tests/unit/nodes/test_orchestrator_no_io.py (1)
tests/helpers/ast_analysis.py (1)
is_docstring(281-311)
🔇 Additional comments (19)
src/omnibase_infra/nodes/node_registration_orchestrator/__init__.py (1)
19-30: LGTM! Clear documentation for heartbeat handling.The Event Handlers section effectively documents the new heartbeat handling capability with a helpful import snippet.
tests/unit/nodes/test_orchestrator_decision_paths.py (1)
529-579: LGTM! Correctly handles direct_handler events.The logic properly skips events with
direct_handler=truefrom workflow pattern validation since they bypass the workflow execution. The updated error message provides clear remediation options.tests/unit/nodes/test_node_registration_orchestrator.py (1)
159-185: LGTM! Test coverage mirrors timeout handling pattern.The test properly validates that heartbeat handling methods are present on the orchestrator class, following the same pattern as the timeout coordinator tests.
src/omnibase_infra/orchestrators/registration/handlers/__init__.py (1)
1-15: LGTM! Standard package initialization.Clean re-export pattern for the heartbeat handler and related types.
src/omnibase_infra/nodes/node_registration_orchestrator/node.py (3)
49-74: LGTM! Clear documentation for heartbeat wiring.The docstring provides a complete example of how to wire the heartbeat handler with the orchestrator.
112-120: LGTM! Proper TYPE_CHECKING imports and initialization.Type hints are properly gated under TYPE_CHECKING to avoid circular dependencies, and the handler attribute is initialized correctly.
Also applies to: 203-203
260-321: LGTM! Heartbeat handling follows timeout coordinator pattern.The implementation correctly mirrors the timeout coordinator pattern with proper delegation, error handling, and comprehensive docstrings. The RuntimeError for missing handler is consistent with the existing timeout coordinator approach.
tests/unit/nodes/test_orchestrator_no_io.py (2)
285-315: LGTM! Correctly whitelists heartbeat delegation methods.The heartbeat handler methods are properly whitelisted as legitimate delegation points that don't violate the pure coordinator principle, following the same pattern as timeout coordination.
487-537: LGTM! Consistent whitelist across test classes.The heartbeat methods whitelist is consistently applied, and the init statement count is appropriately updated to account for the
_heartbeat_handlerinitialization.tests/integration/registration/handlers/conftest.py (1)
97-122: LGTM! Fast window fixture is well-documented.The fast window variant with 5.0 seconds is appropriate for testing deadline calculations without long waits.
tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py (3)
1-66: LGTM! Well-structured test module with comprehensive documentation.The module docstring clearly documents the test coverage areas (happy path, error scenarios, concurrency, etc.) and includes related ticket references. The graceful skip behavior for Docker availability is well-documented for CI/CD pipelines.
73-180: Well-designed test helpers with clear defaults and documentation.The helper functions follow best practices:
- Use keyword-only arguments for clarity (
*parameter)- Proper type annotations with
X | Nonesyntax per coding guidelines- Comprehensive docstrings explaining each parameter
seed_projectionincludes an assertion to fail fast if seeding fails
599-636: Good test for connection error propagation with proper import placement.The test correctly imports
EnumInfraTransportTypeandModelInfraErrorContextwithin the test method to construct the error context. This validates thatInfraConnectionErroris propagated correctly from the projector to the caller, which aligns with the handler's documented error handling behavior.src/omnibase_infra/projectors/projector_registration.py (2)
259-332: LGTM! Upsert SQL correctly updated for last_heartbeat_at field.The SQL changes properly:
- Add
last_heartbeat_atto the INSERT column list (line 262)- Include the corresponding parameter placeholder in VALUES (line 268, $9)
- Update the column in ON CONFLICT DO UPDATE SET (line 278)
- Pass
projection.last_heartbeat_atin the params tuple (line 322)The parameter ordering is correct and the WHERE clause logic for stale update rejection remains unchanged.
717-840: LGTM! Well-implemented update_heartbeat method following established patterns.The new method:
- Follows the same structure as
update_ack_timeout_markerandupdate_liveness_timeout_marker- Properly implements circuit breaker pattern with lock acquisition
- Uses appropriate error types:
InfraConnectionErrorfor connection failures,InfraTimeoutErrorfor cancellations,RuntimeHostErrorfor other errors (per coding guidelines)- Includes comprehensive logging with correlation context
- Sets
updated_at = $3(last_heartbeat_at) which is consistent with the heartbeat update semantics- Docstring includes related ticket reference (OMN-1006)
src/omnibase_infra/orchestrators/registration/handlers/handler_node_heartbeat.py (4)
45-96: LGTM! Well-designed immutable result model.The
ModelHeartbeatHandlerResultmodel:
- Uses
frozen=Truefor immutability and thread safety- Uses
extra="forbid"for strict validation- Uses nullable type annotation
X | Noneper coding guidelines- All fields are well-documented with descriptions
- Follows Pydantic best practices
98-156: LGTM! Handler class follows best practices.The
HandlerNodeHeartbeatclass:
- Uses constructor-based dependency injection for projection_reader and projector
- Is stateless and thread-safe as documented
- Provides a property for
liveness_window_secondsfor configuration visibility- Documents error handling behavior including specific exception types
- Default liveness window (90s) matches the 3x heartbeat interval heuristic
285-305: Exception handling correctly preserves RuntimeHostError subtypes.The exception handling:
- Re-raises all
RuntimeHostErrorsubclasses directly (lines 285-290), preservingInfraConnectionError,InfraTimeoutError, etc.- Only wraps unexpected non-infrastructure errors in
RuntimeHostError(lines 291-305)- Includes detailed logging with error type for debugging
- This addresses the concern from the past review comment about wrapping specific error types
Based on learnings, this follows the guideline to use
InfraConnectionErrorfor connection failures andInfraTimeoutErrorfor operation timeouts.
157-283: LGTM! Handle method implements correct heartbeat processing flow.The method correctly:
- Generates correlation_id if not provided for tracing
- Returns structured result for unknown nodes (node_not_found=True)
- Logs warnings for non-ACTIVE states but still processes the heartbeat (correct for race conditions)
- Calculates liveness_deadline as
event.timestamp + liveness_window- Handles the TOCTOU race condition where entity exists during lookup but is deleted before update
- Returns comprehensive result with previous_state, timestamps, and correlation_id
| # OMN-1006: Node heartbeat events for liveness tracking | ||
| # Heartbeats update last_heartbeat_at and extend liveness_deadline in the | ||
| # registration projection. The HandlerNodeHeartbeat processes these events. | ||
| # Note: direct_handler=true indicates this event bypasses the workflow and is | ||
| # handled by a dedicated handler (HandlerNodeHeartbeat) via handle_heartbeat(). | ||
| - topic: "node.heartbeat" | ||
| event_type: "ModelNodeHeartbeatEvent" | ||
| description: "Periodic heartbeat from active nodes for liveness tracking" | ||
| direct_handler: true |
There was a problem hiding this comment.
Critical: Topic pattern violates ONEX convention.
Line 213 uses "node.heartbeat" which doesn't follow the required ONEX topic pattern used by all other consumed events. This will break topic templating at runtime when the system attempts to replace {env} and {namespace} placeholders.
All consumed events must follow one of these patterns:
{env}.{namespace}.onex.evt.<slug>.v1(external events){env}.{namespace}.onex.internal.<slug>.v1(internal events)
🔎 Proposed fix
- - topic: "node.heartbeat"
+ - topic: "{env}.{namespace}.onex.evt.node-heartbeat.v1"
event_type: "ModelNodeHeartbeatEvent"
description: "Periodic heartbeat from active nodes for liveness tracking"
direct_handler: true📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # OMN-1006: Node heartbeat events for liveness tracking | |
| # Heartbeats update last_heartbeat_at and extend liveness_deadline in the | |
| # registration projection. The HandlerNodeHeartbeat processes these events. | |
| # Note: direct_handler=true indicates this event bypasses the workflow and is | |
| # handled by a dedicated handler (HandlerNodeHeartbeat) via handle_heartbeat(). | |
| - topic: "node.heartbeat" | |
| event_type: "ModelNodeHeartbeatEvent" | |
| description: "Periodic heartbeat from active nodes for liveness tracking" | |
| direct_handler: true | |
| # OMN-1006: Node heartbeat events for liveness tracking | |
| # Heartbeats update last_heartbeat_at and extend liveness_deadline in the | |
| # registration projection. The HandlerNodeHeartbeat processes these events. | |
| # Note: direct_handler=true indicates this event bypasses the workflow and is | |
| # handled by a dedicated handler (HandlerNodeHeartbeat) via handle_heartbeat(). | |
| - topic: "{env}.{namespace}.onex.evt.node-heartbeat.v1" | |
| event_type: "ModelNodeHeartbeatEvent" | |
| description: "Periodic heartbeat from active nodes for liveness tracking" | |
| direct_handler: true |
🤖 Prompt for AI Agents
In src/omnibase_infra/nodes/node_registration_orchestrator/contract.yaml around
lines 208 to 216, the topic "node.heartbeat" violates the ONEX naming convention
and will break runtime templating; replace the topic value with the templated
ONEX pattern (e.g. "{env}.{namespace}.onex.evt.node.heartbeat.v1") so that {env}
and {namespace} can be substituted at runtime, keeping event_type, description
and direct_handler unchanged.
- Fix CRITICAL: Change heartbeat topic to ONEX convention
`node.heartbeat` → `{env}.{namespace}.onex.evt.node-heartbeat.v1`
- Fix CI failure: test_consumed_events_have_topics now passes
- Fix MINOR: Update test docstring consistency for INFRA_MAX_UNIONS
- Fix NITPICK: Add specific timestamp assertion for rapid heartbeat test
MAJOR review item verified correct: Exception handling already preserves
InfraConnectionError/InfraTimeoutError types (they inherit RuntimeHostError).
PR Review - feat(projection): track last_heartbeat_at in registration projection [OMN-1006]Overall AssessmentLGTM with minor observations ✅ This is a well-architected implementation that properly extends the registration projection with heartbeat tracking. The code follows ONEX conventions closely and demonstrates excellent attention to detail in error handling, testing, and documentation. Strengths1. Excellent Error Handling 🎯The heartbeat handler correctly preserves infrastructure error types: except RuntimeHostError:
# Re-raise all infrastructure errors directly (preserves error type)
raiseThis allows callers to catch specific subtypes ( 2. Comprehensive Test Coverage ✅27 integration tests across 8 test classes is outstanding. Tests cover:
3. Strong Type Safety 💪
4. Circuit Breaker Integration 🔒Projector's async with self._circuit_breaker_lock:
await self._check_circuit_breaker("update_heartbeat", corr_id)Follows the caller-held lock pattern from CLAUDE.md. 5. Accurate Timestamp Tracking 📅The PR includes detailed verification comment in
This addresses the core requirement of OMN-1006. Observations1. SQL Schema Migration (Informational)
2. Contract YAML -
|
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py (1)
737-738: Consider using a more specific exception type.The test catches a generic
Exceptionfor the frozen model mutation check. For better precision, you could usepydantic.ValidationErrorsince Pydantic frozen models raise this on mutation attempts.🔎 Suggested improvement
+from pydantic import ValidationError + # ... # Attempt to modify should fail - with pytest.raises(Exception): # ValidationError for frozen models + with pytest.raises(ValidationError): result.success = False # type: ignore[misc]
📜 Review details
Configuration used: defaults
Review profile: CHILL
Plan: Lite
📒 Files selected for processing (3)
src/omnibase_infra/nodes/node_registration_orchestrator/contract.yamltests/integration/registration/handlers/test_handler_node_heartbeat_integration.pytests/unit/validation/test_validator_defaults.py
🚧 Files skipped from review as they are similar to previous changes (2)
- tests/unit/validation/test_validator_defaults.py
- src/omnibase_infra/nodes/node_registration_orchestrator/contract.yaml
🧰 Additional context used
📓 Path-based instructions (1)
**/*.py
📄 CodeRabbit inference engine (CLAUDE.md)
**/*.py: NEVER useAnytypes in Python - Always use specific types, useobjectfor generic dispatchers instead
All data structures must be proper Pydantic models - one model per file
Use nullable type annotationX | None(PEP 604 union syntax) overOptional[X]for null types in Python
All services MUST useModelONEXContainerfor container-based dependency injection
Error classes must raiseOnexErrornot base Exception - useraise OnexError(...) from epattern
Never useisinstancefor protocol resolution - use duck typing through protocols instead
NEVER include passwords, API keys, tokens, secrets, full connection strings with credentials, PII, or private keys in error messages or context
Files:
tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py
🧠 Learnings (1)
📚 Learning: 2025-11-24T16:32:55.606Z
Learnt from: CR
Repo: OmniNode-ai/omninode_bridge PR: 0
File: .cursor/rules/node_standards.mdc:0-0
Timestamp: 2025-11-24T16:32:55.606Z
Learning: Applies to tests/unit/infrastructure/**/test_*.py : All node implementations must have comprehensive unit tests following the testing pattern in `tests/unit/infrastructure/` with tests for node initialization and node execution
Applied to files:
tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py
🔇 Additional comments (10)
tests/integration/registration/handlers/test_handler_node_heartbeat_integration.py (10)
1-66: LGTM!Well-structured module with comprehensive docstring, proper license header, and clean imports. The use of
TYPE_CHECKINGfor type-only imports and the pytestmark configuration for integration tests are appropriate.
68-180: LGTM!Clean test helper factories with:
- Proper keyword-only arguments for clarity
- PEP 604 union syntax (
X | None) per coding guidelines- Sensible defaults for integration testing
- Clear docstrings documenting parameters and return values
187-217: LGTM!Initialization tests correctly verify default and custom liveness window configuration. The redundant assertion on line 202 (
assert handler.liveness_window_seconds == 90.0) after line 201 serves as documentation of the expected default value, which is acceptable.
224-347: LGTM!Comprehensive happy path tests covering:
- Basic heartbeat processing for ACTIVE nodes
- Liveness deadline extension calculations with appropriate tolerance
- Correlation ID preservation and auto-generation
- Selective field updates (ensuring other fields remain unchanged)
Good practice using both result object assertions and direct database state verification.
354-389: LGTM!Well-structured tests for unknown node scenarios, verifying:
node_not_foundflag is set correctly- Error message contains expected context
- Correlation ID is preserved in error responses
396-440: LGTM!Excellent use of parametrization to test all non-ACTIVE states. The test documents the design decision that heartbeats from non-ACTIVE nodes are processed (for tracking) with a warning logged, which handles race conditions during state transitions.
447-523: LGTM!Thorough testing of liveness window deadline calculations:
- Default 90-second window verification
- Event timestamp-based calculation (not current time)
- Progressive deadline extension with consecutive heartbeats
Good test design using past timestamps to verify the calculation is based on event time.
530-596: LGTM!Well-designed concurrency tests that:
- Verify parallel heartbeat processing from multiple nodes
- Acknowledge non-deterministic ordering with concurrent writes (lines 588-595)
- Use appropriate assertion strategy for race conditions (checking final value is one of expected values)
603-713: LGTM!Comprehensive error handling tests covering:
InfraConnectionErrorpropagation (lines 606-642)- Unexpected error wrapping in
RuntimeHostError(lines 644-674)- Race condition handling when entity is deleted between lookup and update (lines 676-712)
Good use of
patch.objectwithAsyncMockfor simulating infrastructure failures.
769-884: LGTM!Excellent database state verification tests that:
- Directly query PostgreSQL to verify column updates
- Ensure
liveness_timeout_emitted_atmarkers are not reset (important for timeout emission logic)- Verify
ack_deadlineremains unchanged (heartbeat should only update heartbeat-specific fields)These tests provide strong guarantees about the atomicity and selectivity of the heartbeat update operation.
Merged OMN-949 DLQ configuration changes with OMN-1006 heartbeat handler. Conflict resolutions: - INFRA_MAX_UNIONS: Combined both changes (606 + 11 = 617, set to 620 with buffer) - Updated threshold history to reflect both merges Changes from main (OMN-949): - DLQ message format and configuration - Topic constants and validation - Error sanitization utilities - DLQ integration tests
Resolves all PR #94 review issues including critical, major, minor, and nitpicks: Exception handling improvements: - Add explicit catches for InfraConnectionError, InfraTimeoutError, InfraAuthenticationError, InfraUnavailableError before generic Exception - Re-raise infrastructure errors directly to preserve error types - Only wrap truly unexpected errors in RuntimeHostError Constant export consistency: - Export DEFAULT_LIVENESS_WINDOW_SECONDS from orchestrators/__init__.py - Replace hardcoded 90.0 values with constant in tests/conftest.py - Update integration tests to use constant for liveness window New timestamp accuracy tests: - test_heartbeat_timestamp_matches_event_timestamp_exactly - test_liveness_deadline_calculation_precision - test_heartbeat_preserves_utc_timezone - test_successive_heartbeats_update_timestamps_monotonically - test_timestamp_accuracy_for_liveness_expired_event_reporting - test_emit_liveness_expiration_includes_last_heartbeat_at - test_emit_liveness_expiration_handles_none_last_heartbeat_at Documentation fixes: - Update test docstring to reflect current union threshold (620)
PR Review: Add last_heartbeat_at Tracking [OMN-1006]SummaryThis PR adds heartbeat timestamp tracking to the registration projection system, enabling accurate liveness expiration reporting. The implementation is well-designed, thoroughly tested, and follows ONEX infrastructure patterns correctly. ✅ Strengths1. Excellent Architecture & Design
2. Outstanding Test Coverage
3. Proper ONEX Patterns
4. Database Design
5. Documentation Quality
🔍 Code Quality ObservationsException Handling (Lines 289-321)EXCELLENT - The error handling pattern is textbook: except (
InfraConnectionError,
InfraTimeoutError,
InfraAuthenticationError,
InfraUnavailableError,
):
raise # Preserve specific error types
except RuntimeHostError:
raise # Preserve other infrastructure errors
except Exception as e:
# Only wrap truly unexpected errors
raise RuntimeHostError(...) from eThis preserves error type fidelity while preventing unexpected exceptions from escaping. Callers can catch specific error types for differentiated handling. Timestamp Handling (Lines 234-236)heartbeat_timestamp = event.timestamp
new_liveness_deadline = heartbeat_timestamp + timedelta(
seconds=self._liveness_window_seconds
)CORRECT - Uses event timestamp directly (accurate source of truth). The SQL Update Method (projector_registration.py:773-813)The
Topic Naming (contract.yaml)- topic: "{env}.{namespace}.onex.evt.node-heartbeat.v1"
event_type: "ModelNodeHeartbeatEvent"
description: "Periodic heartbeat from active nodes for liveness tracking"
direct_handler: truePERFECT - Follows ONEX topic convention: The 🎯 Minor Observations (No Action Required)1. Validation Threshold UpdateINFRA_MAX_UNIONS = 620 # Was 589, increased by 31Rationale Documented: The increase is from merging OMN-949 (DLQ config) + OMN-1006 (heartbeat handler). The comment history shows clear tracking:
This is acceptable as an infrastructure pattern with a documented migration plan to reduce unions via 2. Constant Export HierarchyThe
Good design - This provides flexible import paths while maintaining a single source of truth. Users can import from the most convenient level. 3. Non-ACTIVE Node Handling (Lines 222-228)if not projection.current_state.is_active():
logger.warning(...)
# Still process the heartbeat to update tracking, but log the warningPragmatic choice - Processes heartbeats even for non-ACTIVE nodes to handle race conditions gracefully. This prevents dropped heartbeats during state transitions. 🔒 Security Review✅ Error SanitizationAll error context properly excludes sensitive data:
✅ SQL Injection PreventionAll queries use parameterized statements: await conn.fetchrow(
update_sql,
entity_id, #
domain, #
last_heartbeat_at, #
liveness_deadline, #
)✅ UUID ValidationUUIDs are strongly typed throughout - no string manipulation vulnerabilities. 📊 Performance ConsiderationsDatabase ImpactPOSITIVE:
Heartbeat Write Load:
ConcurrencyThe integration tests verify concurrent heartbeats are handled correctly (TestHandlerNodeHeartbeatConcurrency). PostgreSQL's row-level locking ensures atomic updates. 🧪 Test Quality AnalysisIntegration Test HighlightsFrom 8 test classes, 27+ test methods:
Particularly Strong Tests:
Test Fixture Quality
🎨 Code Style ComplianceONEX Naming Conventions: ✅ PERFECT
Type Annotations: ✅ EXCELLENT
Documentation: ✅ COMPREHENSIVE
🚀 Deployment ReadinessSchema MigrationThe SQL changes are idempotent and backward-compatible:
Rollout Strategy
Zero-downtime deployment: ✅ Possible Monitoring RecommendationsConsider adding metrics for:
📝 Final VerdictOverall Assessment: EXCELLENT ✨This is a high-quality PR that demonstrates:
Recommendation: APPROVE & MERGE 🚢This PR is ready for production deployment. The code quality, test coverage, and adherence to ONEX patterns are exemplary. 🏆 Notable Achievements
Great work on this implementation! 🎉 Additional Notes
Reviewed against ONEX CLAUDE.md standards - Full compliance verified. |
1 participant
Summary
Add
last_heartbeat_atfield to track when the last heartbeat was received from a node, enabling accurate reporting inModelNodeLivenessExpiredevents.Linear Issue: OMN-1006
Changes
model_registration_projection.pylast_heartbeat_at: datetime | Nonefieldschema_registration_projection.sqllast_heartbeat_at TIMESTAMPTZcolumnprojector_registration.pyupdate_heartbeat()methodprojection_reader_registration.pylast_heartbeat_atfrom DB rowstimeout_emitter.pyprojection.last_heartbeat_atinstead ofNonehandler_node_heartbeat.pytest_projection_reader_registration.pyinfra_validators.pyINFRA_MAX_UNIONSthreshold (589 → 600)Acceptance Criteria
ModelRegistrationProjectionhaslast_heartbeat_atfieldlast_heartbeat_atcolumnlast_heartbeat_atviaupdate_heartbeat()ModelNodeLivenessExpiredincludes accuratelast_heartbeat_atTest plan
Follow-up
The heartbeat handler (
handler_node_heartbeat.py) is created but not yet wired into event routing. This can be done in a follow-up ticket when heartbeat event consumption is implemented.Summary by CodeRabbit
New Features
Infrastructure
Tests
Documentation
Chores
✏️ Tip: You can customize this high-level summary in your review settings.