diff --git a/docker/docker-compose.dev-202.yml b/docker/docker-compose.dev-202.yml index 1af813b342..25d75070bd 100644 --- a/docker/docker-compose.dev-202.yml +++ b/docker/docker-compose.dev-202.yml @@ -186,6 +186,12 @@ services: environment: ONEX_RUNTIME_LANE: "compose-dev-202" BIFROST_LANE_OVERLAY_PATH: /app/config/delegation/dev-202.bifrost.yaml + # OMN-20590: this lane's main runtime publishes the runtime tick + # (onex.intent.platform.runtime-tick.v1). Opt-in per lane: the first + # ticks activate every subscribing contract, including the dead-letter + # and consumer-flow prunes (dry_run false). Only the main role acts on + # it, so it is set here and nowhere else in this file. + ONEX_RUNTIME_SCHEDULER_ENABLED: "true" ports: !override - "127.0.0.1:61085:8085" - "172.17.0.1:61085:8085" @@ -211,6 +217,14 @@ services: INFISICAL_CLIENT_ID: ${ONEX_RUNTIME_INFISICAL_CLIENT_ID:-} INFISICAL_CLIENT_SECRET: ${ONEX_RUNTIME_INFISICAL_CLIENT_SECRET:-} INFISICAL_ENVIRONMENT_SLUG: "" + # OMN-20590: this lane publishes the runtime tick (see omninode-runtime + # above), and the two prune effects on this role resolve their local + # archive directory from these names on every tick. Unset, both raise + # KeyError per tick and dead-letter it (read live on this lane + # 2026-10-05). The directories live on this role's own data volume; + # the sink creates them owner-only on first write. + ONEX_DEAD_LETTER_ARCHIVE_DIR: /app/data/dead-letter-archive + ONEX_CONSUMER_FLOW_ARCHIVE_DIR: /app/data/consumer-flow-archive ports: !override - "127.0.0.1:61086:8085" - "172.17.0.1:61086:8085" diff --git a/src/omnibase_infra/runtime/runtime_profile.py b/src/omnibase_infra/runtime/runtime_profile.py index 07a2ceb064..bb1708436a 100644 --- a/src/omnibase_infra/runtime/runtime_profile.py +++ b/src/omnibase_infra/runtime/runtime_profile.py @@ -79,6 +79,12 @@ _SECRET_POLICY_ENV_VAR = "ONEX_SECRET_POLICY" _VALID_PREFETCH_POLICIES = frozenset({"disabled", "best_effort", "required"}) +# OMN-20590: lane-scoped switch for the runtime tick producer. See +# resolve_runtime_scheduler_enabled(). +_RUNTIME_SCHEDULER_ENABLED_ENV_VAR = "ONEX_RUNTIME_SCHEDULER_ENABLED" +_TRUE_SPELLINGS = frozenset({"true", "1", "yes", "on"}) +_FALSE_SPELLINGS = frozenset({"false", "0", "no", "off"}) + class ModelRuntimeProfile(BaseModel): """Schema for a named runtime deployment profile. @@ -316,6 +322,39 @@ def load_runtime_profile(profile_name: str | None = None) -> ModelRuntimeProfile return profile +def resolve_runtime_scheduler_enabled() -> bool: + """Resolve ``ONEX_RUNTIME_SCHEDULER_ENABLED``, the lane's tick-producer switch (OMN-20590). + + Lane-scoped, like ``ONEX_SECRET_POLICY``: a lane overlay sets it once, and + only the ``main`` role acts on it (see + ``runtime_scheduler.start_lane_runtime_scheduler``). It is opt-in + per lane because the first ticks a lane receives activate every contract + subscribed to the runtime tick, including prunes configured to delete. + + Unset or blank is ``False``. A value outside the accepted spellings is + REFUSED rather than read as false: a lane that wrote ``ture`` asked for + ticks, and silently publishing none is the defect this switch exists to + end. + + Returns: + Whether this lane asked for a runtime tick producer. + + Raises: + ProtocolConfigurationError: If the value is not a recognised boolean. + """ + raw = os.environ.get(_RUNTIME_SCHEDULER_ENABLED_ENV_VAR, "").strip().lower() + if not raw: + return False + if raw in _TRUE_SPELLINGS: + return True + if raw in _FALSE_SPELLINGS: + return False + raise ProtocolConfigurationError( + f"{_RUNTIME_SCHEDULER_ENABLED_ENV_VAR}={raw!r} is not a boolean. " + f"Use one of {sorted([*_TRUE_SPELLINGS, *_FALSE_SPELLINGS])}." + ) + + def resolve_secret_resolver_config_path() -> str: """Resolve ``ONEX_SECRET_RESOLVER_CONFIG_PATH`` (OMN-14951). @@ -338,5 +377,6 @@ class of read (a deploy-time-rendered artifact path, "ModelRuntimeProfile", "load_runtime_profile", "resolve_runtime_profile_name", + "resolve_runtime_scheduler_enabled", "resolve_secret_resolver_config_path", ] diff --git a/src/omnibase_infra/runtime/runtime_scheduler.py b/src/omnibase_infra/runtime/runtime_scheduler.py index 02fb78de62..4e229dac8a 100644 --- a/src/omnibase_infra/runtime/runtime_scheduler.py +++ b/src/omnibase_infra/runtime/runtime_scheduler.py @@ -76,6 +76,7 @@ from redis.exceptions import RedisError from redis.exceptions import TimeoutError as RedisTimeoutError +from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope from omnibase_infra.enums import EnumInfraTransportType from omnibase_infra.errors import ( InfraConnectionError, @@ -85,19 +86,23 @@ ModelTimeoutErrorContext, ProtocolConfigurationError, ) -from omnibase_infra.event_bus.event_bus_kafka import EventBusKafka -from omnibase_infra.event_bus.models import ModelEventHeaders from omnibase_infra.mixins import MixinAsyncCircuitBreaker +from omnibase_infra.protocols import ProtocolEventBusLike from omnibase_infra.runtime.enums import EnumSchedulerStatus from omnibase_infra.runtime.models import ( ModelRuntimeSchedulerConfig, ModelRuntimeSchedulerMetrics, ModelRuntimeTick, ) +from omnibase_infra.runtime.runtime_profile import resolve_runtime_scheduler_enabled from omnibase_infra.utils.util_error_sanitization import sanitize_error_string logger = logging.getLogger(__name__) +# The event_type every runtime-tick subscriber routes on (handler_routing +# ``event_type: platform.runtime-tick`` in each subscribing contract). +RUNTIME_TICK_EVENT_TYPE = "platform.runtime-tick" + class RuntimeScheduler(MixinAsyncCircuitBreaker): """Runtime scheduler that emits RuntimeTick events at configured intervals. @@ -145,13 +150,14 @@ class RuntimeScheduler(MixinAsyncCircuitBreaker): def __init__( self, config: ModelRuntimeSchedulerConfig, - event_bus: EventBusKafka, + event_bus: ProtocolEventBusLike, ) -> None: """Initialize the RuntimeScheduler. Args: config: Configuration model containing all scheduler settings. - event_bus: EventBusKafka instance for publishing tick events. + event_bus: Any event bus with ``publish_envelope`` (Kafka in a lane, + in-memory in tests and local runs). Raises: ProtocolConfigurationError: If config or event_bus is None. @@ -418,18 +424,20 @@ async def emit_tick(self, now: datetime | None = None) -> None: await self._record_tick_failure(correlation_id) raise - # Create headers for the event - headers = ModelEventHeaders( + # OMN-20590: the tick travels as a ModelEventEnvelope carrying the + # event_type every subscribing contract routes on. A bare tick is + # refused outright by EventBusSubcontractWiring._deserialize_to_envelope + # (no ``payload`` key), and at the auto-wiring consume boundary it is + # wrapped in a synthesized envelope whose minted id has no lineage + # (OMN-18958), so the envelope is the one shape both paths take whole. + envelope: ModelEventEnvelope[ModelRuntimeTick] = ModelEventEnvelope( + payload=tick, correlation_id=correlation_id, - message_id=tick_id, - timestamp=tick_time, - source=f"runtime-scheduler.{self.scheduler_id}", - event_type="runtime.tick.v1", + event_type=RUNTIME_TICK_EVENT_TYPE, + source_tool=f"runtime-scheduler.{self.scheduler_id}", + tenant_id=None, ) - # Serialize tick to JSON bytes - tick_bytes = tick.model_dump_json().encode("utf-8") - # Prepare error context for ONEX error types ctx = ModelInfraErrorContext( transport_type=EnumInfraTransportType.KAFKA, @@ -440,11 +448,10 @@ async def emit_tick(self, now: datetime | None = None) -> None: try: # Publish tick event - await self._event_bus.publish( + await self._event_bus.publish_envelope( + envelope=envelope, topic=self._config.tick_topic, key=self.scheduler_id.encode("utf-8"), - value=tick_bytes, - headers=headers, ) # Record success @@ -1067,4 +1074,81 @@ async def _persist_sequence_number(self) -> None: await self._close_valkey_client() -__all__: list[str] = ["RuntimeScheduler"] +# ============================================================================= +# Lane wiring (OMN-20590) +# ============================================================================= +# +# RuntimeScheduler was built (OMN-953) and never started by any runtime, so +# nothing published the runtime tick on any lane while seven contracts +# subscribed to it (read live 2026-10-05: the .201 dev and stability brokers +# and the k3s onex-lab broker held HIGH-WATERMARK 0 on the topic). +# +# Ownership: a lane runs one ``main`` runtime and several secondary roles that +# often share one compose environment anchor. Two producers on one bus would +# double every tick, so only the ``main`` role publishes. +# +# Opt-in per lane: the first ticks a lane receives activate every subscribing +# contract, two of which prune rows with ``dry_run: false``. A lane turns its +# producer on in the overlay that owns it, so a redeploy keeps the setting and +# a lane nobody edited does not change. A ``main`` runtime that was not asked +# says so at WARNING on every boot, naming the topic. + +# The one role that owns the lane's tick producer. +TICK_PRODUCER_PROFILE = "main" + + +async def start_lane_runtime_scheduler( + event_bus: ProtocolEventBusLike, + runtime_profile: str, +) -> RuntimeScheduler | None: + """Start the runtime tick producer when this process owns it. + + Args: + event_bus: The kernel's started event bus. + runtime_profile: The validated ROLE identity of this process + (``resolve_runtime_profile_name()``). + + Returns: + The running scheduler, which the caller must ``stop()`` at shutdown, + or ``None`` when this process does not publish ticks. + + Raises: + ProtocolConfigurationError: If the lane's switch holds a value that + is not a boolean. + """ + enabled = resolve_runtime_scheduler_enabled() + if runtime_profile != TICK_PRODUCER_PROFILE: + logger.debug( + "Runtime tick producer not started: profile %r does not own it (owner: %r)", + runtime_profile, + TICK_PRODUCER_PROFILE, + ) + return None + config = ModelRuntimeSchedulerConfig.default() + if not enabled: + logger.warning( + "Runtime tick producer NOT started on this lane: " + "ONEX_RUNTIME_SCHEDULER_ENABLED is not true, so nothing publishes " + "%s and every contract subscribed to it (prunes, reapers, timeout " + "evaluation, pollers) never fires. Set it in the lane overlay to " + "turn the producer on.", + config.tick_topic, + ) + return None + scheduler = RuntimeScheduler(config=config, event_bus=event_bus) + await scheduler.start() + logger.info( + "Runtime tick producer started: scheduler_id=%s topic=%s interval_ms=%d", + scheduler.scheduler_id, + config.tick_topic, + config.tick_interval_ms, + ) + return scheduler + + +__all__: list[str] = [ + "RUNTIME_TICK_EVENT_TYPE", + "TICK_PRODUCER_PROFILE", + "RuntimeScheduler", + "start_lane_runtime_scheduler", +] diff --git a/src/omnibase_infra/runtime/service_kernel.py b/src/omnibase_infra/runtime/service_kernel.py index b3acbe5bac..2f62ed005c 100644 --- a/src/omnibase_infra/runtime/service_kernel.py +++ b/src/omnibase_infra/runtime/service_kernel.py @@ -196,6 +196,10 @@ resolve_runtime_profile_name, resolve_secret_resolver_config_path, ) +from omnibase_infra.runtime.runtime_scheduler import ( + RuntimeScheduler, + start_lane_runtime_scheduler, +) from omnibase_infra.runtime.util_container_wiring import ( wire_infrastructure_services, ) @@ -1656,6 +1660,8 @@ async def bootstrap() -> int: plugin_unsubscribe_callbacks: list[Callable[[], Awaitable[None]]] = [] # Contract registry unsubscribe functions and router (separate domain) contract_router: ContractRegistrationEventRouter | None = None + # OMN-20590: the lane's runtime tick producer (main profile, opt-in). + runtime_tick_scheduler: RuntimeScheduler | None = None contract_unsub_registered: Callable[[], Awaitable[None]] | None = None contract_unsub_deregistered: Callable[[], Awaitable[None]] | None = None contract_unsub_heartbeat: Callable[[], Awaitable[None]] | None = None @@ -4872,6 +4878,16 @@ async def _wiring_health_loop() -> None: }, ) + # 9.6b. Start the lane's runtime tick producer (OMN-20590). + # RuntimeScheduler (OMN-953) was never started by any runtime, so + # nothing published onex.intent.platform.runtime-tick.v1 and every + # contract subscribed to it never fired. Only the main role publishes, + # and only on a lane whose overlay sets ONEX_RUNTIME_SCHEDULER_ENABLED. + runtime_tick_scheduler = await start_lane_runtime_scheduler( + event_bus, + resolve_runtime_profile_name(), + ) + # 9.7. Start runtime error triage consumer (OMN-5655) # Subscribes to runtime-error events and routes them to the # HandlerRuntimeErrorTriage for first-match-wins triage processing. @@ -5152,6 +5168,13 @@ async def _triage_on_message( ) plugin_unsubscribe_callbacks.clear() + # Stop the runtime tick producer (OMN-20590). stop() is graceful and + # idempotent: it bounds the tick task's wait and degrades a failed + # sequence persist to a warning itself. + if runtime_tick_scheduler is not None: + await runtime_tick_scheduler.stop() + runtime_tick_scheduler = None + # Stop contract registry router and consumers if contract_router is not None: try: @@ -5456,6 +5479,10 @@ async def _triage_on_message( correlation_id, ) + # Cleanup the runtime tick producer (OMN-20590); stop() is idempotent. + if runtime_tick_scheduler is not None: + await runtime_tick_scheduler.stop() + # Cleanup contract registry router and consumers if contract_router is not None: try: diff --git a/tests/integration/runtime/test_runtime_tick_producer_golden_chain_omn20590.py b/tests/integration/runtime/test_runtime_tick_producer_golden_chain_omn20590.py new file mode 100644 index 0000000000..e1c9bfaf7c --- /dev/null +++ b/tests/integration/runtime/test_runtime_tick_producer_golden_chain_omn20590.py @@ -0,0 +1,140 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""OMN-20590: a tick from the lane's own producer reaches a tick-driven handler. + +The chain: ``start_lane_runtime_scheduler`` (main profile, lane switch on) -> +``RuntimeScheduler.emit_tick`` -> event bus -> auto-wired runtime-tick +subscription -> MessageDispatchEngine -> HandlerGitHubApiPoll -> result topic. + +RED on the parent: no lane producer could be started at all (no +``start_lane_runtime_scheduler``, no lane switch), so no tick ever reached a +handler. The hand-built envelope in ``test_github_api_poll_golden_chain`` proves +the consumer half; this proves the producer the lanes actually run. +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope +from omnibase_infra.event_bus.event_bus_inmemory import EventBusInmemory +from omnibase_infra.event_bus.models.model_event_message import ModelEventMessage +from omnibase_infra.nodes.node_github_pr_poller_effect.handlers import ( + handler_github_api_poll, +) +from omnibase_infra.nodes.node_github_pr_poller_effect.models.model_github_poller_config import ( + ModelGitHubPollerConfig, +) +from omnibase_infra.nodes.node_github_pr_poller_effect.models.model_github_poller_result import ( + ModelGitHubPollerResult, +) +from omnibase_infra.runtime.auto_wiring import discover_contracts_from_paths +from omnibase_infra.runtime.auto_wiring.handler_wiring import wire_from_manifest +from omnibase_infra.runtime.message_dispatch_engine import MessageDispatchEngine +from omnibase_infra.runtime.runtime_scheduler import start_lane_runtime_scheduler + +_OUTPUT_TOPIC = "onex.evt.github.pr-status.v1" +_REPO = "OmniNode-ai/omnibase_infra" + + +def _contract_path() -> Path: + return ( + Path(__file__).resolve().parents[3] + / "src" + / "omnibase_infra" + / "nodes" + / "node_github_pr_poller_effect" + / "contract.yaml" + ) + + +class _FakeGitHubClient: + def __init__(self, **_kwargs: object) -> None: + pass + + def fetch_open_prs_for_triage(self, repo: str) -> list[dict[str, object]]: + assert repo == _REPO + return [ + { + "number": 20590, + "title": "Runtime tick producer", + "draft": False, + "labels": [], + "updated_at": datetime.now(tz=UTC).isoformat().replace("+00:00", "Z"), + "combined_status": "success", + "review_states": ["APPROVED"], + } + ] + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_lane_producer_tick_drives_a_tick_subscriber( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ONEX_RUNTIME_SCHEDULER_ENABLED", "true") + monkeypatch.setenv("ONEX_RUNTIME_SCHEDULER_PERSIST_SEQUENCE", "false") + # Long interval: the test drives one tick by hand, the loop adds none. + monkeypatch.setenv("ONEX_RUNTIME_SCHEDULER_TICK_INTERVAL_MS", "60000") + monkeypatch.setenv("GITHUB_TOKEN", "test-token") + monkeypatch.setattr( + handler_github_api_poll, + "_load_contract_config", + lambda: ModelGitHubPollerConfig( + repos=[_REPO], + poll_interval_seconds=10, + stale_threshold_hours=48, + ), + ) + monkeypatch.setattr(handler_github_api_poll, "GitHubHttpClient", _FakeGitHubClient) + + manifest = discover_contracts_from_paths([_contract_path()]) + assert manifest.total_errors == 0 + + bus = EventBusInmemory(environment="test", group="runtime-tick-producer-chain") + await bus.start() + try: + observed: asyncio.Queue[ModelEventEnvelope[ModelGitHubPollerResult]] = ( + asyncio.Queue() + ) + + async def collect_result(message: ModelEventMessage) -> None: + await observed.put( + ModelEventEnvelope[ModelGitHubPollerResult].model_validate_json( + message.value + ) + ) + + await bus.subscribe( + topic=_OUTPUT_TOPIC, + group_id="runtime-tick-producer-result-collector", + on_message=collect_result, + ) + + engine = MessageDispatchEngine() + report = await wire_from_manifest( + manifest, + engine, + event_bus=bus, + environment="test", + ) + assert report.total_wired == 1 + engine.freeze() + + scheduler = await start_lane_runtime_scheduler(bus, "main") + assert scheduler is not None + try: + await scheduler.emit_tick() + result = await asyncio.wait_for(observed.get(), timeout=5) + finally: + await scheduler.stop() + + assert result.payload.errors == [] + assert result.payload.repos_polled == [_REPO] + assert result.payload.prs_polled == 1 + finally: + await bus.close() diff --git a/tests/unit/runtime/test_runtime_scheduler.py b/tests/unit/runtime/test_runtime_scheduler.py index 8a366bf23a..f9891a037e 100644 --- a/tests/unit/runtime/test_runtime_scheduler.py +++ b/tests/unit/runtime/test_runtime_scheduler.py @@ -35,12 +35,13 @@ import os from collections.abc import Awaitable, Coroutine from datetime import UTC, datetime -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, _Call, patch from uuid import UUID, uuid4 import pytest from pydantic import ValidationError +from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope from omnibase_infra.errors import InfraUnavailableError, ProtocolConfigurationError from omnibase_infra.protocols import ProtocolEventBusLike from omnibase_infra.runtime import runtime_scheduler as runtime_scheduler_module @@ -50,7 +51,10 @@ ModelRuntimeSchedulerMetrics, ModelRuntimeTick, ) -from omnibase_infra.runtime.runtime_scheduler import RuntimeScheduler +from omnibase_infra.runtime.runtime_scheduler import ( + RUNTIME_TICK_EVENT_TYPE, + RuntimeScheduler, +) from omnibase_infra.topics import SUFFIX_RUNTIME_TICK # Upper bound for waiting on the running tick loop. Ticks are expected every @@ -89,6 +93,14 @@ async def _until_emitted() -> ModelRuntimeSchedulerMetrics: ) +def _published_tick(call: _Call) -> ModelRuntimeTick: + """The tick carried by one recorded ``publish_envelope`` call (OMN-20590).""" + envelope = call.kwargs["envelope"] + assert isinstance(envelope, ModelEventEnvelope) + assert envelope.event_type == RUNTIME_TICK_EVENT_TYPE + return ModelRuntimeTick.model_validate(envelope.payload) + + class _RecordingAsyncio: """Stand-in for the scheduler module's ``asyncio`` that records waits. @@ -128,7 +140,7 @@ def mock_event_bus() -> AsyncMock: AsyncMock configured to simulate EventBusKafka behavior. """ bus = AsyncMock(spec=ProtocolEventBusLike) - bus.publish = AsyncMock(return_value=None) + bus.publish_envelope = AsyncMock(return_value=None) bus.start = AsyncMock(return_value=None) bus.stop = AsyncMock(return_value=None) return bus @@ -691,17 +703,41 @@ async def test_emit_tick_creates_valid_tick( await scheduler.emit_tick() # Verify publish was called - mock_event_bus.publish.assert_called_once() + mock_event_bus.publish_envelope.assert_called_once() # Get the call arguments - call_args = mock_event_bus.publish.call_args + call_args = mock_event_bus.publish_envelope.call_args assert call_args.kwargs["topic"] == "test.runtime.tick.v1" assert call_args.kwargs["key"] == b"test-scheduler" - # Verify tick content (value is JSON bytes) - tick_bytes = call_args.kwargs["value"] - assert b"test-scheduler" in tick_bytes - assert b"sequence_number" in tick_bytes + # Verify tick content (OMN-20590: the tick travels as an envelope) + tick = _published_tick(call_args) + assert tick.scheduler_id == "test-scheduler" + assert tick.sequence_number == 1 + + async def test_tick_published_as_envelope_the_consumers_accept( + self, scheduler: RuntimeScheduler, mock_event_bus: AsyncMock + ) -> None: + """OMN-20590: the tick is published as an envelope, not a bare model. + + ``EventBusSubcontractWiring._deserialize_to_envelope`` refuses a bare + tick (no ``payload`` key), the auto-wiring boundary wraps one in an + envelope with a minted, lineage-less id, and every subscribing contract + routes on ``platform.runtime-tick``. RED on the parent: ``emit_tick`` + called ``publish`` with the bare tick's JSON bytes. + """ + import json + + await scheduler.emit_tick() + + mock_event_bus.publish.assert_not_called() + call_args = mock_event_bus.publish_envelope.call_args + envelope = call_args.kwargs["envelope"] + wire = json.loads(json.dumps(envelope.model_dump(mode="json"))) + decoded = ModelEventEnvelope[object].model_validate(wire) + assert decoded.event_type == RUNTIME_TICK_EVENT_TYPE + tick = ModelRuntimeTick.model_validate(decoded.payload) + assert tick.scheduler_id == "test-scheduler" async def test_emit_tick_increments_sequence( self, scheduler: RuntimeScheduler, mock_event_bus: AsyncMock @@ -727,11 +763,9 @@ async def test_emit_tick_with_injected_time( await scheduler.emit_tick(now=fixed_time) # Verify the tick contains the injected time - call_args = mock_event_bus.publish.call_args - tick_bytes = call_args.kwargs["value"] - - # The JSON should contain the fixed time - assert b"2025-06-15" in tick_bytes + call_args = mock_event_bus.publish_envelope.call_args + # The tick should carry the injected time + assert _published_tick(call_args).now == fixed_time async def test_tick_published_to_kafka( self, scheduler: RuntimeScheduler, mock_event_bus: AsyncMock @@ -742,7 +776,7 @@ async def test_tick_published_to_kafka( await scheduler.emit_tick() # Verify 3 publishes occurred - assert mock_event_bus.publish.call_count == 3 + assert mock_event_bus.publish_envelope.call_count == 3 async def test_tick_has_unique_ids( self, scheduler: RuntimeScheduler, mock_event_bus: AsyncMock @@ -752,18 +786,15 @@ async def test_tick_has_unique_ids( await scheduler.emit_tick() # Get both tick payloads - call1 = mock_event_bus.publish.call_args_list[0] - call2 = mock_event_bus.publish.call_args_list[1] + call1 = mock_event_bus.publish_envelope.call_args_list[0] + call2 = mock_event_bus.publish_envelope.call_args_list[1] - # Parse the JSON to extract IDs - import json - - tick1 = json.loads(call1.kwargs["value"]) - tick2 = json.loads(call2.kwargs["value"]) + tick1 = _published_tick(call1) + tick2 = _published_tick(call2) # IDs should be unique - assert tick1["tick_id"] != tick2["tick_id"] - assert tick1["correlation_id"] != tick2["correlation_id"] + assert tick1.tick_id != tick2.tick_id + assert tick1.correlation_id != tick2.correlation_id async def test_tick_loop_emits_at_interval( self, @@ -783,7 +814,7 @@ async def test_tick_loop_emits_at_interval( await _wait_for_ticks(scheduler, minimum=2) await scheduler.stop() - published = mock_event_bus.publish.call_count + published = mock_event_bus.publish_envelope.call_count assert published >= 2 # Every published tick was preceded by one interval wait, and every @@ -902,12 +933,9 @@ async def test_concurrent_tick_emission_sequence_safety( assert scheduler.current_sequence_number == 20 # Verify each tick has correct sequence in published data - import json - sequences = [] - for call in mock_event_bus.publish.call_args_list: - tick_data = json.loads(call.kwargs["value"]) - sequences.append(tick_data["sequence_number"]) + for call in mock_event_bus.publish_envelope.call_args_list: + sequences.append(_published_tick(call).sequence_number) # All sequences should be unique assert len(set(sequences)) == 20 @@ -948,7 +976,9 @@ async def test_metrics_track_failures( from omnibase_infra.errors import InfraConnectionError # Configure event bus to fail - mock_event_bus.publish = AsyncMock(side_effect=Exception("Publish failed")) + mock_event_bus.publish_envelope = AsyncMock( + side_effect=Exception("Publish failed") + ) scheduler = RuntimeScheduler(config=scheduler_config, event_bus=mock_event_bus) @@ -1028,7 +1058,9 @@ async def test_circuit_breaker_opens_on_failures( ) -> None: """Test that circuit breaker opens after threshold failures.""" # Configure to fail - mock_event_bus.publish = AsyncMock(side_effect=Exception("Network error")) + mock_event_bus.publish_envelope = AsyncMock( + side_effect=Exception("Network error") + ) scheduler = RuntimeScheduler(config=scheduler_config, event_bus=mock_event_bus) @@ -1050,7 +1082,9 @@ async def test_circuit_breaker_blocks_when_open( ) -> None: """Test that InfraUnavailableError is raised when circuit is open.""" # Configure to fail - mock_event_bus.publish = AsyncMock(side_effect=Exception("Network error")) + mock_event_bus.publish_envelope = AsyncMock( + side_effect=Exception("Network error") + ) scheduler = RuntimeScheduler(config=scheduler_config, event_bus=mock_event_bus) @@ -1100,7 +1134,9 @@ async def test_circuit_breaker_auto_reset_after_timeout( ) # Start with failures to open circuit - mock_event_bus.publish = AsyncMock(side_effect=Exception("Network error")) + mock_event_bus.publish_envelope = AsyncMock( + side_effect=Exception("Network error") + ) scheduler = RuntimeScheduler(config=config, event_bus=mock_event_bus) # Open the circuit @@ -1116,7 +1152,7 @@ async def test_circuit_breaker_auto_reset_after_timeout( await asyncio.sleep(1.1) # Now make publish succeed - mock_event_bus.publish = AsyncMock(return_value=None) + mock_event_bus.publish_envelope = AsyncMock(return_value=None) # Should be able to emit (circuit in half-open state) await scheduler.emit_tick() @@ -1131,7 +1167,9 @@ async def test_start_blocked_when_circuit_open( ) -> None: """Test that start() raises when circuit breaker is already open.""" # Configure to fail - mock_event_bus.publish = AsyncMock(side_effect=Exception("Network error")) + mock_event_bus.publish_envelope = AsyncMock( + side_effect=Exception("Network error") + ) scheduler = RuntimeScheduler(config=scheduler_config, event_bus=mock_event_bus) @@ -1237,7 +1275,7 @@ async def intermittent_failure(*args: object, **kwargs: object) -> None: if fail_count <= 2: raise RuntimeError("Intermittent failure") - mock_event_bus.publish = AsyncMock(side_effect=intermittent_failure) + mock_event_bus.publish_envelope = AsyncMock(side_effect=intermittent_failure) scheduler = RuntimeScheduler(config=scheduler_config, event_bus=mock_event_bus) @@ -1249,7 +1287,7 @@ async def intermittent_failure(*args: object, **kwargs: object) -> None: await scheduler.stop() # Should have attempted multiple publishes despite failures - assert mock_event_bus.publish.call_count >= 3 + assert mock_event_bus.publish_envelope.call_count >= 3 async def test_scheduler_handles_rapid_start_stop( self, scheduler: RuntimeScheduler @@ -1351,9 +1389,9 @@ async def test_tick_headers_contain_correlation_id( """Test that tick events include correlation ID in headers.""" await scheduler.emit_tick() - call_args = mock_event_bus.publish.call_args - headers = call_args.kwargs["headers"] + call_args = mock_event_bus.publish_envelope.call_args + envelope = call_args.kwargs["envelope"] - # Headers should have correlation_id - assert headers.correlation_id is not None - assert isinstance(headers.correlation_id, UUID) + # The envelope carries the tick's correlation_id + assert isinstance(envelope.correlation_id, UUID) + assert envelope.correlation_id == envelope.payload.correlation_id diff --git a/tests/unit/runtime/test_runtime_scheduler_wiring.py b/tests/unit/runtime/test_runtime_scheduler_wiring.py new file mode 100644 index 0000000000..1a090363ab --- /dev/null +++ b/tests/unit/runtime/test_runtime_scheduler_wiring.py @@ -0,0 +1,200 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""OMN-20590: the lane's runtime tick producer is started by the kernel. + +RED on the parent: ``RuntimeScheduler`` had no instantiation site in ``src``, +so no lane published ``onex.intent.platform.runtime-tick.v1``, and there was no +lane switch to turn one on. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock + +import pytest +import yaml + +from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope +from omnibase_infra.errors import ProtocolConfigurationError +from omnibase_infra.protocols import ProtocolEventBusLike +from omnibase_infra.runtime.models import ModelRuntimeTick +from omnibase_infra.runtime.runtime_profile import resolve_runtime_scheduler_enabled +from omnibase_infra.runtime.runtime_scheduler import ( + RUNTIME_TICK_EVENT_TYPE, + start_lane_runtime_scheduler, +) +from omnibase_infra.topics import SUFFIX_RUNTIME_TICK + +pytestmark = pytest.mark.unit + +_FLAG = "ONEX_RUNTIME_SCHEDULER_ENABLED" +_DOCKER_DIR = Path(__file__).resolve().parents[3] / "docker" + + +@pytest.fixture +def bus() -> AsyncMock: + return AsyncMock(spec=ProtocolEventBusLike) + + +@pytest.fixture(autouse=True) +def _no_valkey(monkeypatch: pytest.MonkeyPatch) -> None: + # No sequence persistence in unit tests: the scheduler would try Valkey. + monkeypatch.setenv("ONEX_RUNTIME_SCHEDULER_PERSIST_SEQUENCE", "false") + monkeypatch.setenv("ONEX_RUNTIME_SCHEDULER_TICK_INTERVAL_MS", "60000") + + +class TestFlag: + @pytest.mark.parametrize("raw", ["", " "]) + def test_unset_or_blank_is_off( + self, monkeypatch: pytest.MonkeyPatch, raw: str + ) -> None: + monkeypatch.setenv(_FLAG, raw) + assert resolve_runtime_scheduler_enabled() is False + + def test_absent_is_off(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(_FLAG, raising=False) + assert resolve_runtime_scheduler_enabled() is False + + @pytest.mark.parametrize("raw", ["true", "TRUE", "1", "yes", "on"]) + def test_true_spellings(self, monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + monkeypatch.setenv(_FLAG, raw) + assert resolve_runtime_scheduler_enabled() is True + + @pytest.mark.parametrize("raw", ["false", "0", "no", "off"]) + def test_false_spellings(self, monkeypatch: pytest.MonkeyPatch, raw: str) -> None: + monkeypatch.setenv(_FLAG, raw) + assert resolve_runtime_scheduler_enabled() is False + + def test_unparseable_value_is_refused( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv(_FLAG, "ture") + with pytest.raises(ProtocolConfigurationError, match=_FLAG): + resolve_runtime_scheduler_enabled() + + +@pytest.mark.asyncio +class TestStartLaneRuntimeScheduler: + async def test_main_profile_with_flag_starts_and_publishes_envelope( + self, monkeypatch: pytest.MonkeyPatch, bus: AsyncMock + ) -> None: + monkeypatch.setenv(_FLAG, "true") + scheduler = await start_lane_runtime_scheduler(bus, "main") + assert scheduler is not None + try: + assert scheduler.is_running + await scheduler.emit_tick() + finally: + await scheduler.stop() + + call = bus.publish_envelope.call_args + assert call.kwargs["topic"] == SUFFIX_RUNTIME_TICK + envelope = call.kwargs["envelope"] + assert envelope.event_type == RUNTIME_TICK_EVENT_TYPE + assert isinstance(envelope.payload, ModelRuntimeTick) + + @pytest.mark.parametrize("profile", ["effects", "workers", "projection-api"]) + async def test_secondary_role_never_starts_one( + self, monkeypatch: pytest.MonkeyPatch, bus: AsyncMock, profile: str + ) -> None: + monkeypatch.setenv(_FLAG, "true") + assert await start_lane_runtime_scheduler(bus, profile) is None + bus.publish_envelope.assert_not_called() + + async def test_main_profile_without_flag_starts_none_and_says_so( + self, + monkeypatch: pytest.MonkeyPatch, + bus: AsyncMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + monkeypatch.delenv(_FLAG, raising=False) + with caplog.at_level(logging.WARNING): + assert await start_lane_runtime_scheduler(bus, "main") is None + bus.publish_envelope.assert_not_called() + assert any( + SUFFIX_RUNTIME_TICK in r.getMessage() and _FLAG in r.getMessage() + for r in caplog.records + if r.levelno == logging.WARNING + ) + + async def test_unparseable_flag_refuses_boot( + self, monkeypatch: pytest.MonkeyPatch, bus: AsyncMock + ) -> None: + monkeypatch.setenv(_FLAG, "ture") + with pytest.raises(ProtocolConfigurationError): + await start_lane_runtime_scheduler(bus, "main") + + +def _load_compose(name: str) -> dict[str, Any]: + # `!override` is a compose-only tag SafeLoader refuses. + text = (_DOCKER_DIR / name).read_text(encoding="utf-8") + document = yaml.safe_load(text.replace("!override", "")) + assert isinstance(document, dict) + return document + + +def _service_env(document: dict[str, Any], service: str) -> dict[str, str]: + env = (document.get("services") or {}).get(service, {}).get("environment") or {} + if isinstance(env, list): + return dict(item.split("=", 1) for item in env if "=" in item) + return {str(k): str(v) for k, v in env.items()} + + +class TestLaneOverlays: + def test_overlay_dev_202_turns_on_the_main_runtime_producer(self) -> None: + document = _load_compose("docker-compose.dev-202.yml") + assert _service_env(document, "omninode-runtime").get(_FLAG) == "true" + + def test_overlay_dev_202_gives_the_tick_driven_prunes_their_archive( + self, + ) -> None: + # The two prune effects on runtime-effects resolve these on every tick; + # an opted-in lane without them dead-letters every tick (read live on + # dev-202, 2026-10-05: KeyError 'ONEX_CONSUMER_FLOW_ARCHIVE_DIR'). + env = _service_env( + _load_compose("docker-compose.dev-202.yml"), "runtime-effects" + ) + for name in ("ONEX_DEAD_LETTER_ARCHIVE_DIR", "ONEX_CONSUMER_FLOW_ARCHIVE_DIR"): + assert env.get(name, "").startswith("/app/data/"), name + + def test_overlay_flag_is_set_by_dev_202_alone(self) -> None: + # Every other lane changes only by an explicit overlay edit; this test + # names the lanes that opted in so the next one is a reviewed diff. + opted_in = sorted( + path.name + for path in _DOCKER_DIR.glob("docker-compose*.yml") + if _FLAG in path.read_text(encoding="utf-8") + ) + assert opted_in == ["docker-compose.dev-202.yml"] + + +def test_envelope_survives_the_consumer_deserializer_shape() -> None: + """The bytes a Kafka bus would publish validate as the consumer expects.""" + from datetime import UTC, datetime + from uuid import uuid4 + + now = datetime.now(UTC) + tick = ModelRuntimeTick( + now=now, + tick_id=uuid4(), + sequence_number=1, + scheduled_at=now, + correlation_id=uuid4(), + scheduler_id="runtime-scheduler-default", + tick_interval_ms=1000, + ) + envelope = ModelEventEnvelope( + payload=tick, + correlation_id=tick.correlation_id, + event_type=RUNTIME_TICK_EVENT_TYPE, + source_tool="runtime-scheduler.runtime-scheduler-default", + tenant_id=None, + ) + wire = json.loads(json.dumps(envelope.model_dump(mode="json"))) + decoded = ModelEventEnvelope[object].model_validate(wire) + assert decoded.event_type == RUNTIME_TICK_EVENT_TYPE + assert ModelRuntimeTick.model_validate(decoded.payload) == tick