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
3 changes: 0 additions & 3 deletions config/validation/handler_dispatch_entrypoint_baseline.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -43,9 +43,6 @@ known_entrypointless:
- contract: node_auth_gate_compute
handler: HandlerAuthGate
module: omnibase_infra.nodes.node_auth_gate_compute.handlers.handler_auth_gate
- contract: node_broker_disk_watermark_compute
handler: HandlerBrokerDiskWatermark
module: omnibase_infra.nodes.node_broker_disk_watermark_compute.handlers.handler_broker_disk_watermark
- contract: node_chain_verify_reducer
handler: HandlerChainVerify
module: omnibase_infra.nodes.node_chain_verify_reducer.handlers.handler_chain_verify
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
{
"base_ref": "71ab1bf6deb4252edd9593f547dd5a4b4e60939b",
"canonical_handler_module": "omnibase_infra.nodes.node_broker_disk_watermark_compute.handlers.handler_broker_disk_watermark",
"legacy_handler_module": "omnibase_infra.nodes.node_broker_disk_watermark_compute.handlers.handler_broker_disk_watermark",
"node_id": "omnibase_infra.nodes.node_broker_disk_watermark_compute",
"parity": {
"selected_input_hashes": [
"sha256:a717499da9403e40025b12891003a8849b9106e79d6cec5e8c20757852d2a2d2",
"sha256:5e6f7ee2fa56cbec23c8dc73be4019b28cb43c129f47828518da9ca6503cffb8",
"sha256:170122f07cbb1949c0a230f71ec3facd9847f5d2b4e33e9cb5c8e550209dc836",
"sha256:a60cefb436747aca50e48a192df8f4d5002c6a2bbefdd5279324f8c0c9cb1b55",
"sha256:3f29ef12329a6a7d1e6d25ef5e5d669465f12ce2fc7ae11f87aa4390b903350a"
],
"status": "pass",
"test_ids": [
"tests/unit/nodes/test_broker_disk_watermark_dispatch_entrypoint_omn14816.py::test_handler_exposes_handle_entrypoint",
"tests/unit/nodes/test_broker_disk_watermark_dispatch_entrypoint_omn14816.py::test_real_dispatch_callback_returns_p0_output",
"tests/unit/nodes/test_broker_disk_watermark_dispatch_entrypoint_omn14816.py::test_real_dispatch_callback_returns_clean_output"
]
},
"preserved_symbols": [
"_classify",
"_real_docker_info",
"_real_disk_usage",
"_resolve_volume_mountpoint",
"HandlerBrokerDiskWatermark._probe_path"
],
"receipt_schema": "handflip_proof.v1"
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
{
"branch_coverage_pct": 42.86,
"candidate_count": 6,
"coverage_target": 80.0,
"handler_module": "omnibase_infra.nodes.node_broker_disk_watermark_compute.handlers.handler_broker_disk_watermark",
"handler_module_sha256": "sha256:a1d3f32eaac23f505ea084a87218f8480f2ba6430bcfc674d83d4c413353a48e",
"meets_target": false,
"node_id": "omnibase_infra.nodes.node_broker_disk_watermark_compute",
"receipt_schema": "adequacy_receipt.v1",
"recorded_at": "2026-07-20T02:51:57.262774+00:00",
"recorder_schema": "adequacy_receipt.v1",
"selected_count": 5,
"selected_input_hashes": [
"sha256:a717499da9403e40025b12891003a8849b9106e79d6cec5e8c20757852d2a2d2",
"sha256:5e6f7ee2fa56cbec23c8dc73be4019b28cb43c129f47828518da9ca6503cffb8",
"sha256:170122f07cbb1949c0a230f71ec3facd9847f5d2b4e33e9cb5c8e550209dc836",
"sha256:a60cefb436747aca50e48a192df8f4d5002c6a2bbefdd5279324f8c0c9cb1b55",
"sha256:3f29ef12329a6a7d1e6d25ef5e5d669465f12ce2fc7ae11f87aa4390b903350a"
],
"uncovered_waiver": {
"reason": "OMN-14816 hand-flips node_broker_disk_watermark_compute to canonical def-B via a PURE RENAME probe_disk_watermark->handle (OMN-14781 path): the probe body is preserved byte-identically and the business-logic helpers (_classify / _probe_path / _real_docker_info / _real_disk_usage / _resolve_volume_mountpoint) are re-derived byte-identical base_ref<->HEAD by the ratchet from git. Equivalence is proven by that git-verbatim preservation plus the green runtime-dispatch parity tests over the selected input corpus, not by a broad branch-coverage claim over the legacy handler body; the real docker/shutil subprocess helpers (_real_docker_info / _real_disk_usage / _resolve_volume_mountpoint) are network/daemon-dependent and are exercised by the existing patched unit suite, so they are stubbed in the deterministic dispatch corpus.",
"uncovered_branch_count": 9
},
"volatile_mask": []
}
1 change: 0 additions & 1 deletion scripts/ci/canonical_handler_shape_baseline.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
"omnibase_infra.nodes.node_artifact_change_detector_effect",
"omnibase_infra.nodes.node_artifact_reconciliation_orchestrator",
"omnibase_infra.nodes.node_auth_gate_compute",
"omnibase_infra.nodes.node_broker_disk_watermark_compute",
"omnibase_infra.nodes.node_build_loop_write_effect",
"omnibase_infra.nodes.node_bus_forwarder_effect",
"omnibase_infra.nodes.node_chain_orchestrator",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,23 @@
# Related Tickets:
# - OMN-13009: P0 Redpanda broker storage exhaustion (error system:28)
# - OMN-13008: Automated worktree GC + .201 docker/disk GC + disk watermark alert ratchet
# - OMN-14816: HandlerBrokerDiskWatermark now exposes the canonical def-B
# handle() dispatch entrypoint via a behavior-preserving hand-flip from
# probe_disk_watermark(); business-logic helpers preserved byte-identically
# (canonical-shape ratchet OMN-14355, hand-flip proof path OMN-14781).
# Runtime auto-wiring uses handle_async() to execute blocking docker/disk
# probes off the event loop while preserving the typed handle() contract.
name: "node_broker_disk_watermark_compute"
contract_name: "node_broker_disk_watermark_compute"
node_name: "node_broker_disk_watermark_compute"
contract_version:
major: 1
minor: 0
patch: 0
patch: 2
node_version:
major: 1
minor: 0
patch: 0
patch: 1
node_type: "COMPUTE_GENERIC"
description: >
Compute node that probes the docker data-root filesystem (via docker info -> DockerRootDir, NOT host /) and per-broker-volume disk usage, then classifies each against warn (85%) and p0 (95%) watermarks. Also cross-checks that remaining free bytes exceed redpanda storage_min_free_bytes (default 10 MiB). Returns a list of ModelBrokerDiskCheck records — one per volume plus one for the data-root — each with severity CLEAN / WARN / P0. A downstream orchestrator creates Linear tickets on P0 and Slack alerts on WARN. docker info and shutil.disk_usage are injected seams for testability.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,16 @@
# Copyright (c) 2026 OmniNode Team
"""HandlerBrokerDiskWatermark — probes docker data-root + broker volumes.

Canonical definition-B handler (OMN-14355 / OMN-14816): the dispatch entrypoint
is the single ``handle(request) -> response`` method that auto-wiring's
``_make_dispatch_callback`` binds. It takes the typed ``ModelBrokerDiskWatermarkInput``
and returns ``ModelBrokerDiskWatermarkOutput`` — no event-envelope import, no
``Plugin*`` base, stateless/deterministic compute. This is a behavior-preserving
hand-flip (OMN-14781) of the former ``probe_disk_watermark`` op-method: the method
was renamed to ``handle`` with its body and every business-logic helper
(``_classify`` / ``_probe_path`` / ``_real_docker_info`` / ``_real_disk_usage`` /
``_resolve_volume_mountpoint``) preserved byte-identically.

Comment thread
coderabbitai[bot] marked this conversation as resolved.
Algorithm
---------
1. Resolve docker data-root via ``docker info --format '{{json .}}'`` →
Expand All @@ -23,6 +33,7 @@

from __future__ import annotations

import asyncio
import json
import logging
import shutil
Expand Down Expand Up @@ -164,11 +175,14 @@ def handler_category(self) -> EnumHandlerTypeCategory:
"""Behavioral classification: pure compute (no external I/O side-effects)."""
return EnumHandlerTypeCategory.COMPUTE

def probe_disk_watermark(
def handle(
self, inp: ModelBrokerDiskWatermarkInput
) -> ModelBrokerDiskWatermarkOutput:
"""Probe docker data-root + broker volumes and classify severity.

Canonical def-B dispatch entrypoint (renamed from ``probe_disk_watermark``
with the body preserved byte-identically; OMN-14816 hand-flip).

Args:
inp: Watermark probe configuration.

Expand Down Expand Up @@ -291,5 +305,17 @@ def _probe_path(
below_min_free_floor=below_floor,
)

async def handle_async(
self, inp: ModelBrokerDiskWatermarkInput
) -> ModelBrokerDiskWatermarkOutput:
"""Run the blocking probe implementation off the event loop.

Auto-wiring prefers ``handle_async`` over ``handle`` when both exist, so
runtime dispatch avoids executing docker subprocess and disk-usage calls
inline on the event loop while preserving the synchronous handler
contract for direct callers and tests.
"""
return await asyncio.to_thread(self.handle, inp)


__all__ = ["HandlerBrokerDiskWatermark"]
6 changes: 6 additions & 0 deletions src/omnibase_infra/validation/validation_exemptions.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -2671,6 +2671,12 @@ pattern_exemptions:
'handle' is the canonical ONEX def-B dispatch entrypoint name that auto-wiring's _make_dispatch_callback binds; a handler exposing neither handle nor handle_async is bound to _missing_handle and raises ModelOnexError on every dispatch (OMN-14510 missing-handle burn-down). HandlerLlmCliSubprocess was regenerated to def-B under OMN-14804 to add this entrypoint, matching the ProtocolMessageHandler convention used by every other node handler on node_llm_inference_effect.

ticket: OMN-14804
- file_pattern: 'handler_broker_disk_watermark\.py'
violation_pattern: "Function name 'handle' is too generic"
reason: >
'handle' is the canonical ONEX handler method name (definition B, OMN-14355) that auto-wiring's _make_dispatch_callback binds. OMN-14816 hand-flipped this node's op-method probe_disk_watermark() to the canonical handle() dispatch entrypoint; the name is protocol-defined and renaming it would break the canonical shape this flip establishes. Mirrors the handler_release_identity handle exemption.

ticket: OMN-14816
- file_pattern: 'evaluator_invariant\.py'
class_pattern: "Class 'InvariantEvaluator'"
violation_pattern: 'has \d+ methods'
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,179 @@
# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 OmniNode Team
"""OMN-14816 — HandlerBrokerDiskWatermark is REACHABLE through the real dispatch path.

This is the RED-against-EXISTS-but-WRONG proof for the canonical def-B flip.

Before OMN-14816 the handler was contract-declared, wired, ingress-valid and CI-green
while exposing only the op-method ``probe_disk_watermark()`` and NO ``handle``.
Auto-wiring's ``_make_dispatch_callback`` looks for ``handle_async`` then ``handle``;
finding neither it binds ``_missing_handle``, which raises::

ModelOnexError: Auto-wired handler HandlerBrokerDiskWatermark does not expose a
callable handle() or handle_async() dispatch entrypoint.

...on the FIRST dispatch. These tests drive the REAL production dispatch callback over
the REAL handler class (no fake handler, no patched entrypoint), so they FAIL against
the entrypoint-less pre-flip handler and PASS only once the canonical def-B ``handle``
exists. The node uses ``operation_match`` (no contract ``event_model``), so the callback
signature-introspects ``handle(inp: ModelBrokerDiskWatermarkInput)`` and coerces the
payload into that model at the adapter boundary (OMN-14716 parity with runtime_local).

Ticket: OMN-14816
"""

from __future__ import annotations

import asyncio
from uuid import uuid4

import pytest

from omnibase_core.models.events.model_event_envelope import ModelEventEnvelope
from omnibase_infra.enums import EnumDispatchStatus
from omnibase_infra.models.dispatch.model_dispatch_result import ModelDispatchResult
from omnibase_infra.nodes.node_broker_disk_watermark_compute.handlers.handler_broker_disk_watermark import (
HandlerBrokerDiskWatermark,
)
from omnibase_infra.nodes.node_broker_disk_watermark_compute.models.enum_disk_severity import (
EnumDiskSeverity,
)
from omnibase_infra.nodes.node_broker_disk_watermark_compute.models.model_broker_disk_watermark_input import (
ModelBrokerDiskWatermarkInput,
)
from omnibase_infra.nodes.node_broker_disk_watermark_compute.models.model_broker_disk_watermark_output import (
ModelBrokerDiskWatermarkOutput,
)
from omnibase_infra.runtime.auto_wiring.handler_wiring import _make_dispatch_callback

_GiB = 1024**3
_DOCKER_INFO = {"DockerRootDir": "/data"}


def _p0_disk_usage(path: str) -> tuple[int, int, int]:
"""96% used — above the p0 watermark (deterministic; no real docker/shutil)."""
total = 100 * _GiB
used = int(0.96 * total)
return total, used, total - used


def _healthy_disk_usage(path: str) -> tuple[int, int, int]:
"""70% used — well below warn (deterministic)."""
total = 100 * _GiB
used = 70 * _GiB
return total, used, total - used


def _input(cid: object, disk_usage: object) -> ModelBrokerDiskWatermarkInput:
return ModelBrokerDiskWatermarkInput(
correlation_id=cid, # type: ignore[arg-type]
docker_info_runner=lambda: _DOCKER_INFO,
disk_usage_runner=disk_usage, # type: ignore[arg-type]
)


@pytest.mark.unit
def test_handler_exposes_handle_entrypoint() -> None:
"""The bare invariant: auto-wiring can only bind handle/handle_async.

RED against the pre-OMN-14816 handler, which exposed only
``probe_disk_watermark``.
"""
assert callable(getattr(HandlerBrokerDiskWatermark, "handle", None)), (
"HandlerBrokerDiskWatermark exposes no handle(); auto-wiring binds "
"_missing_handle and every dispatch raises ModelOnexError."
)
assert callable(getattr(HandlerBrokerDiskWatermark, "handle_async", None)), (
"HandlerBrokerDiskWatermark must expose handle_async(); auto-wiring runs "
"the synchronous docker/disk probes through the event-loop dispatch path."
)


@pytest.mark.unit
@pytest.mark.asyncio
async def test_real_dispatch_callback_returns_p0_output() -> None:
"""A real payload dispatched through the REAL callback reaches handle() and
returns a P0 classification.

Against the entrypoint-less handler this raises ModelOnexError (_missing_handle)
rather than returning a result — that raise IS the bug, caught here.
"""
cid = uuid4()
callback = _make_dispatch_callback(HandlerBrokerDiskWatermark())
envelope: ModelEventEnvelope[object] = ModelEventEnvelope(
payload=_input(cid, _p0_disk_usage),
correlation_id=cid,
event_type="ModelBrokerDiskWatermarkInput",
)

result = await callback(envelope)

assert result is not None, "Dispatch produced no result — handle() never ran."
assert isinstance(result, ModelDispatchResult)
assert result.status is EnumDispatchStatus.SUCCESS
assert len(result.output_events) == 1, (
f"Expected exactly one ModelBrokerDiskWatermarkOutput; got "
f"{result.output_events!r}"
)
out = result.output_events[0]
assert isinstance(out, ModelBrokerDiskWatermarkOutput)
assert out.correlation_id == cid
assert out.max_severity is EnumDiskSeverity.P0
assert "docker-data-root" in out.p0_labels
assert out.docker_root_dir == "/data"


@pytest.mark.unit
@pytest.mark.asyncio
async def test_real_dispatch_callback_returns_clean_output() -> None:
"""A healthy filesystem dispatched through the REAL callback returns CLEAN."""
cid = uuid4()
callback = _make_dispatch_callback(HandlerBrokerDiskWatermark())
envelope: ModelEventEnvelope[object] = ModelEventEnvelope(
payload=_input(cid, _healthy_disk_usage),
correlation_id=cid,
event_type="ModelBrokerDiskWatermarkInput",
)

result = await callback(envelope)

assert result is not None
assert isinstance(result, ModelDispatchResult)
out = result.output_events[0]
assert isinstance(out, ModelBrokerDiskWatermarkOutput)
assert out.correlation_id == cid
assert out.max_severity is EnumDiskSeverity.CLEAN
assert out.p0_labels == ()
assert out.warn_labels == ()


@pytest.mark.unit
@pytest.mark.asyncio
async def test_real_dispatch_callback_runs_blocking_probe_in_worker_thread(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The runtime callback binds handle_async, which delegates handle via to_thread."""
cid = uuid4()
delegated: list[object] = []

async def _record_to_thread(
func: object, /, *args: object, **kwargs: object
) -> object:
delegated.append(func)
return func(*args, **kwargs) # type: ignore[misc]

monkeypatch.setattr(asyncio, "to_thread", _record_to_thread)

handler = HandlerBrokerDiskWatermark()
callback = _make_dispatch_callback(handler)
envelope: ModelEventEnvelope[object] = ModelEventEnvelope(
payload=_input(cid, _healthy_disk_usage),
correlation_id=cid,
event_type="ModelBrokerDiskWatermarkInput",
)

result = await callback(envelope)

assert result is not None
assert delegated == [handler.handle]
Loading
Loading