Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
fb931c8
feat(kafka): implement contract-driven topic configuration [OMN-881]
jonahgabriel Dec 18, 2025
f7299a6
fix(validators): reduce params with config model, fix contract versio…
jonahgabriel Dec 18, 2025
3cc3964
docs(kafka): address PR #54 review feedback [OMN-881]
jonahgabriel Dec 18, 2025
1867977
feat(introspection): add async cache lock for thread safety [OMN-881]
jonahgabriel Dec 18, 2025
ba8c160
fix(introspection): address PR #54 review feedback [OMN-881]
jonahgabriel Dec 18, 2025
ce148d4
fix(introspection): address PR #54 review feedback round 2 [OMN-881]
jonahgabriel Dec 18, 2025
81353e6
feat(introspection): add topic validation, class defaults, and integr…
jonahgabriel Dec 18, 2025
e8596e0
feat(introspection): add version suffix validation for topic names [O…
jonahgabriel Dec 18, 2025
debfd67
test(introspection): add comprehensive test coverage for PR #54 revie…
jonahgabriel Dec 19, 2025
3d38b10
fix(introspection): address PR #54 review feedback round 3 [OMN-881]
jonahgabriel Dec 19, 2025
417a715
Merge origin/main into feature branch
jonahgabriel Dec 24, 2025
2a4ea48
fix(introspection): address PR review feedback [OMN-881]
jonahgabriel Dec 24, 2025
729dbfd
fix(introspection): address all PR #54 review feedback [OMN-881]
jonahgabriel Dec 24, 2025
c499947
fix(ci): resolve formatting and introspection test failures [OMN-881]
jonahgabriel Dec 24, 2025
ba1455a
Merge origin/main into feature branch
jonahgabriel Dec 24, 2025
183de06
fix(introspection): address PR #54 review feedback and fix CI [OMN-881]
jonahgabriel Dec 24, 2025
ed784dc
fix(tests): address PR #54 release-ready review feedback [OMN-881]
jonahgabriel Dec 25, 2025
afb2ee7
fix(tests): address final PR #54 review comments [OMN-881]
jonahgabriel Dec 25, 2025
e89bad2
fix(tests): address PR #54 release-ready nitpicks and polish [OMN-881]
jonahgabriel Dec 25, 2025
db1b342
Merge origin/main into feature branch
jonahgabriel Dec 25, 2025
ef48654
fix(introspection): address PR #54 review - version suffix and metric…
jonahgabriel Dec 25, 2025
1435023
Merge origin/main - resolve conflicts and add performance metrics to …
jonahgabriel Dec 25, 2025
451dbf6
fix(services): update timeout_emitter import to use ProtocolEventBus
jonahgabriel Dec 25, 2025
65cc5ac
fix(introspection): address PR #54 review - type safety, tests, docs …
jonahgabriel Dec 25, 2025
47e52ac
fix(introspection): address PR #54 review - CI, tests, docs, robustne…
jonahgabriel Dec 25, 2025
82c90b8
Merge origin/main - resolve conflicts and update union limit
jonahgabriel Dec 25, 2025
27c00f9
fix(typing): use ProtocolEventBus instead of object for event_bus field
jonahgabriel Dec 25, 2025
48f55a1
fix(pr-review): address PR #54 review feedback - type safety, race co…
jonahgabriel Dec 25, 2025
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 88 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,72 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

### Breaking Changes

> **IMPORTANT**: This section documents API changes that may require code modifications when upgrading. Review each item carefully before upgrading.

#### MixinNodeIntrospection API (OMN-881, PR #54)

##### 1. Cache Invalidation Method Signature Change

**`invalidate_introspection_cache()` is now synchronous (was async)**

This is a **breaking change** for any code that awaits this method.

| Aspect | Details |
|--------|---------|
| **What changed** | Method signature changed from `async def` to `def` (synchronous) |
| **Why it changed** | Cache invalidation is a simple in-memory operation (setting `_introspection_cache = None`) that does not require async I/O. Synchronous semantics simplify usage and avoid unnecessary coroutine overhead. |
| **Error if not migrated** | `TypeError: object NoneType can't be used in 'await' expression` |

**Migration Steps**:

```python
# BEFORE (will cause TypeError after upgrade)
await node.invalidate_introspection_cache()

# AFTER (correct usage)
node.invalidate_introspection_cache()
```

**Search pattern** to find affected code:
```bash
grep -r "await.*invalidate_introspection_cache" --include="*.py"
```

##### 2. Configuration Model API

**`initialize_introspection()` requires `ModelIntrospectionConfig`**

The initialization method uses a typed configuration model for all parameters.

| Aspect | Details |
|--------|---------|
| **What changed** | `initialize_introspection(config: ModelIntrospectionConfig)` is the initialization API |
| **Why** | Typed configuration model provides validation, IDE support, and extensibility |
| **Model location** | `omnibase_infra.models.discovery.ModelIntrospectionConfig` |

**Usage Example**:

```python
from uuid import uuid4
from omnibase_infra.models.discovery import ModelIntrospectionConfig
from omnibase_infra.mixins import MixinNodeIntrospection

class MyNode(MixinNodeIntrospection):
def __init__(self, event_bus=None):
config = ModelIntrospectionConfig(
node_id=uuid4(),
node_type="EFFECT",
event_bus=event_bus,
version="1.0.0",
cache_ttl=300.0,
)
self.initialize_introspection(config)

async def shutdown(self):
# Note: invalidate_introspection_cache() is now SYNC (see above)
self.invalidate_introspection_cache()
```

#### Error Code for Unhandled node_kind (OMN-990, PR #73)
- **Error code changed from `VALIDATION_ERROR` to `INTERNAL_ERROR`**: When `DispatchContextEnforcer.create_context_for_dispatcher()` encounters an unhandled `node_kind` value, it now raises `ModelOnexError` with `INTERNAL_ERROR` instead of `VALIDATION_ERROR`.
- **Old**: `error_code=EnumCoreErrorCode.VALIDATION_ERROR`
Expand All @@ -25,6 +91,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

### Added

#### Node Introspection (OMN-881, PR #54)
- **ModelIntrospectionConfig**: Configuration model for `MixinNodeIntrospection` that provides typed configuration
- `node_id` (required): Unique identifier for this node instance (UUID)
- `node_type` (required): Type of node (EFFECT, COMPUTE, REDUCER, ORCHESTRATOR). Cannot be empty (min_length=1).
- `event_bus`: Optional event bus for publishing introspection and heartbeat events. Uses duck typing (`object | None`) to accept any object implementing `ProtocolEventBus` protocol.
- `version`: Node version string (default: `"1.0.0"`)
- `cache_ttl`: Cache time-to-live in seconds (default: `300.0`, minimum: `0.0`)
- `operation_keywords`: Optional set of keywords to identify operation methods. If None, uses `MixinNodeIntrospection.DEFAULT_OPERATION_KEYWORDS`.
- `exclude_prefixes`: Optional set of prefixes to exclude from capability discovery. If None, uses `MixinNodeIntrospection.DEFAULT_EXCLUDE_PREFIXES`.
- `introspection_topic`: Topic for publishing introspection events (default: `"node.introspection"`). ONEX topics (starting with `onex.`) require version suffix (e.g., `.v1`).
- `heartbeat_topic`: Topic for publishing heartbeat events (default: `"node.heartbeat"`). ONEX topics require version suffix.
- `request_introspection_topic`: Topic for receiving introspection requests (default: `"node.request_introspection"`). ONEX topics require version suffix.
- Model is frozen and forbids extra fields for immutability and strict validation.
Comment on lines +94 to +106

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Update event_bus field description to match the actual type.

The changelog describes event_bus as using duck typing with object | None, but other PR context indicates it is now explicitly typed as a ProtocolEventBus (breaking change for callers).

To avoid confusion for upgraders, please update this bullet to reflect the real signature (e.g., ProtocolEventBus | None) and drop the object | None wording so the documented API matches the implementation.

🤖 Prompt for AI Agents
In CHANGELOG.md around lines 94 to 106, the entry for ModelIntrospectionConfig
incorrectly describes the event_bus field as duck-typed "object | None"; update
that bullet to state the actual type signature used by the codebase (e.g.,
"ProtocolEventBus | None") and remove the "object | None" wording so the
changelog reflects the breaking-change typing, keeping the rest of the sentence
about accepting an object implementing the protocol only if you want to clarify
compatibility.

- **Performance Metrics Tracking**:
- Added `IntrospectionPerformanceMetrics` dataclass (internal) and `ModelIntrospectionPerformanceMetrics` Pydantic model (for event payloads)
- Added `get_performance_metrics()` method for monitoring introspection operation timing and threshold violations
- Performance thresholds: `get_capabilities` <50ms, `discover_capabilities` <30ms, `total_introspection` <50ms, `cache_hit` <1ms
- **Topic Default Constants**: Exported constants for default topic names:
- `DEFAULT_INTROSPECTION_TOPIC = "node.introspection"`
- `DEFAULT_HEARTBEAT_TOPIC = "node.heartbeat"`
- `DEFAULT_REQUEST_INTROSPECTION_TOPIC = "node.request_introspection"`

#### Handlers
- **HttpHandler** (OMN-237, PR #26): HTTP REST protocol handler for MVP
- GET and POST operations using httpx async client
Expand Down
56 changes: 56 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -1166,8 +1166,64 @@ capabilities = node.get_capabilities()
5. Monitor introspection topic consumer groups for unexpected consumers
6. Use network segmentation to isolate introspection traffic if required

**Cache Operations**:
The mixin manages introspection cache with TTL-based invalidation:

- **Cache Variables**: `_introspection_cache`, `_introspection_cached_at`, `_introspection_cache_ttl`
- **Cache Methods**:
- `get_introspection_data()` - Returns cached data if TTL not expired, otherwise refreshes
- `invalidate_introspection_cache()` - Clears cache to force refresh on next call (synchronous)

**Thread Safety Considerations**:

The `MixinNodeIntrospection` is designed for **single-threaded asyncio usage** and does NOT provide internal thread synchronization. Cache operations require understanding the concurrency model for safe usage.

**Instance-Level Cache** (`_introspection_cache`, `_introspection_cached_at`):
- Cache operations are **synchronous** (no async locking)
- Safe for cooperative asyncio concurrency (single event loop)
- **NOT safe** for multi-threaded access without external synchronization
- `invalidate_introspection_cache()` does not acquire any locks
- If called concurrently with `get_introspection_data()`, cache state may be inconsistent

**Class-Level Cache** (`_class_method_cache`):
- Shared `ClassVar` across all instances of the same class
- Population uses check-then-set pattern (not atomic)
- **Benign race condition**: Multiple threads may populate the cache simultaneously, but all produce identical results since method signatures are immutable after class definition
- Use `_invalidate_class_method_cache()` to clear if dynamic method registration occurs

**Background Tasks** (heartbeat loop, registry listener):
- Run as asyncio tasks within the event loop
- Designed for cooperative async concurrency
- Share access to instance state without locking
- **NOT safe** to call task methods from multiple threads

**Multi-Threaded Usage Pattern**:

If using this mixin in a multi-threaded application (e.g., with `concurrent.futures.ThreadPoolExecutor`), external synchronization is required:

```python
import threading

class ThreadSafeNode(MixinNodeIntrospection):
def __init__(self, config):
self._introspection_lock = threading.Lock()
self.initialize_introspection(...)

def invalidate_introspection_cache(self) -> None:
with self._introspection_lock:
super().invalidate_introspection_cache()

async def get_introspection_data(self):
# Note: async lock needed for async methods
# Consider asyncio.Lock() for async coordination
return await super().get_introspection_data()
```

**Recommendation**: For high-concurrency async environments, prefer using a single asyncio event loop with cooperative multitasking rather than multi-threading.

**Related**:
- Implementation: `src/omnibase_infra/mixins/mixin_node_introspection.py`
- Thread Safety Pattern: `docs/architecture/CIRCUIT_BREAKER_THREAD_SAFETY.md` (similar pattern)
- Ticket: OMN-893
- See `MixinNodeIntrospection.get_capabilities()` for filtering logic details

Expand Down
Loading