Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
14 changes: 14 additions & 0 deletions docker/docker-compose.dev-202.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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"
Expand Down
40 changes: 40 additions & 0 deletions src/omnibase_infra/runtime/runtime_profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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).

Expand All @@ -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",
]
118 changes: 101 additions & 17 deletions src/omnibase_infra/runtime/runtime_scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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",
]
27 changes: 27 additions & 0 deletions src/omnibase_infra/runtime/service_kernel.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
Loading
Loading