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
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ build-backend = "hatchling.build"

[project]
name = "omnibase_core"
version = "0.46.6"
version = "0.46.7"
description = "ONEX Core Framework - Base classes and essential implementations"
authors = [
{name = "OmniNode.ai", email = "contact@omninode.ai"}
Expand Down
35 changes: 35 additions & 0 deletions src/omnibase_core/contracts/runtime_ops_verb_allowlist.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
# SPDX-License-Identifier: MIT
---
# src/omnibase_core/contracts/runtime_ops_verb_allowlist.yaml
# Governed data (NOT code) for the RUNTIME_OPS_READBACK no-PR evidence class
# (OMN-14168). Mirrors the data-not-code posture of
# scripts/ci/test_selection_adjacency.yaml.
#
# A RUNTIME_OPS readback receipt (ModelDodReceipt.evidence_class == RUNTIME_OPS)
# describes an independently-verified, no-source-change runtime-ops fix
# (kubectl patch, live restart, config repair) that produced NO repo diff and
# NO PR. Its ``mutation_verb`` MUST be one of the verbs below. The allowlist is
# deliberately bounded to live-runtime mutations that cannot edit desired-state
# SOURCE: ``git`` and image-digest promotion are intentionally EXCLUDED — a
# source change produces a diff and therefore a PR, which must route through the
# normal merged-PR gate (CONTRACT_CITES_MERGE_COMMIT), not this class (G2).
#
# Both enforcement surfaces read THIS file as the single source of truth:
# * omnibase_core ModelDodReceipt.mutation_verb field validator, and
# * omnimarket DurableEvidenceGate RUNTIME_OPS_READBACK branch (G2b).
#
# To add or remove a verb, edit this file (governed change) — never hardcode the
# set in Python.

schema_version: 1

# Bounded runtime-ops mutation verbs. Each is a live mutation of a *running*
# runtime object (deployment/pod/config), never an edit of source-of-record.
runtime_ops_verbs:
- patch # kubectl patch / strategic merge on a live Deployment
- rollout # kubectl rollout restart / rollout undo
- scale # kubectl scale (replica count on a live workload)
- restart # live container/service restart
- recreate # docker compose up --force-recreate of a running lane
- config-repair # live config/secret repair applied to a running workload
11 changes: 11 additions & 0 deletions src/omnibase_core/enums/governance/enum_evidence_class.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,11 +34,22 @@ class to be proven by a Playwright run with a proxy-origin network
UNCLASSIFIED
No UI/backend distinction asserted or inferred. Treated as not
UI-class — the proxy-origin requirement does not apply.
RUNTIME_OPS
Evidence for a no-source-change runtime-ops fix (OMN-14168) — a
``kubectl patch``, live restart, or config repair with NO repo diff and
NO PR. A receipt of this class carries the runtime-ops mutation block
(``mutation_command`` / ``mutation_verb`` / ``target_identity`` /
``no_source_change`` / ``prevention_followup``) and is proven by a
read-only live readback, not a merged PR. The DurableEvidenceGate keys
its RUNTIME_OPS_READBACK branch on this class; a receipt of this class
must NOT carry a ``pr_number`` (work with a PR routes through the normal
merged-PR gate instead).
"""

UI_DASHBOARD = "ui_dashboard"
BACKEND = "backend"
UNCLASSIFIED = "unclassified"
RUNTIME_OPS = "runtime_ops"


__all__ = ["EnumEvidenceClass"]
148 changes: 146 additions & 2 deletions src/omnibase_core/models/contracts/ticket/model_dod_receipt.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,12 @@

# check_types whose proof requires captured stdout. Listed here (not in the
# validator body) so the policy is grep-able and easy to extend.
EXECUTABLE_CHECK_TYPES = frozenset({"command", "test_passes", "endpoint", "grep"})
# ``runtime_readback`` (OMN-14168) is the RUNTIME_OPS class's read-only live
# probe — an empty readback is indistinguishable from "never ran", so it is
# held to the same non-empty-stdout bar as the other executable proofs (G3).
EXECUTABLE_CHECK_TYPES = frozenset(
{"command", "test_passes", "endpoint", "grep", "runtime_readback"}
)

# check_types whose proof is structurally weak (file presence only). The
# transition policy demotes these to ADVISORY regardless of verifier identity.
Expand Down Expand Up @@ -290,7 +295,66 @@ class ModelDodReceipt(BaseModel):
"UI_DASHBOARD the receipt-gate requires a Playwright proxy-origin "
"network trace and rejects direct curl/wget/httpx probes. When set "
"to BACKEND a direct HTTP probe is accepted (opt-out of the UI gate). "
"None lets the gate infer the class from the probe_command."
"When set to RUNTIME_OPS (OMN-14168) the receipt asserts a no-source-"
"change runtime-ops fix and MUST carry the runtime-ops mutation block "
"below with no ``pr_number``. None lets the gate infer the class from "
"the probe_command."
),
)
mutation_command: str | None = Field(
default=None,
max_length=10000,
description=(
"RUNTIME_OPS only (OMN-14168): the exact live-ops command that was "
"applied (e.g. 'kubectl -n onex-dev patch deployment omnidash "
"--type strategic ...', a restart, or a config repair). Distinct from "
"``probe_command`` which is the read-only readback. Required when "
"``evidence_class == RUNTIME_OPS``."
),
)
mutation_verb: str | None = Field(
default=None,
description=(
"RUNTIME_OPS only (OMN-14168): the parsed verb of ``mutation_command``. "
"Must be a member of the GOVERNED runtime-ops verb allowlist "
"(contracts/runtime_ops_verb_allowlist.yaml). ``git`` and image-digest "
"promotion are intentionally excluded — a source change produces a "
"diff and therefore a PR, which routes through the normal merged-PR "
"gate instead of this class. Required when "
"``evidence_class == RUNTIME_OPS``."
),
)
target_identity: str | None = Field(
default=None,
max_length=2000,
description=(
"RUNTIME_OPS only (OMN-14168): the mutated runtime object "
"(lane / namespace / deployment / pod / node id) the readback was "
"taken against, e.g. 'onex-dev/Deployment/omnidash'. Re-read live at "
"gate time (Surface B) to reject a stale/replayed readback. Required "
"when ``evidence_class == RUNTIME_OPS``."
),
)
no_source_change: bool = Field(
default=False,
description=(
"RUNTIME_OPS only (OMN-14168): asserts the fix produced NO repo diff "
"and NO PR. MUST be True when ``evidence_class == RUNTIME_OPS`` — a "
"runtime-ops close waives only the merged-PR requirement, and only for "
"genuine no-source-change ops. A source change has a diff and routes "
"through the normal PR gate."
),
)
prevention_followup: str | None = Field(
default=None,
max_length=2000,
description=(
"RUNTIME_OPS only (OMN-14168, G5 recurrence ratchet): a linked "
"recurrence-prevention follow-up (a GitOps/declarative guardrail "
"ticket id or PR URL). Required when ``evidence_class == RUNTIME_OPS`` "
"so the no-PR-ops hatch cannot become a comfortable permanent bypass — "
"the hatch is for emergency runtime recovery, not steady-state no-PR "
"ops. Mirrors the DEFECT_PREVENTION_GATE repair-to-ratchet rule."
),
)

Expand Down Expand Up @@ -379,6 +443,41 @@ def _validate_schema_version(cls, v: str) -> str:
)
return v

@field_validator("mutation_verb")
@classmethod
def _validate_mutation_verb(cls, v: str | None) -> str | None:
"""Reject any ``mutation_verb`` outside the governed runtime-ops allowlist.

The allowlist is GOVERNED DATA
(``contracts/runtime_ops_verb_allowlist.yaml``), not hardcoded here, so
adding/removing a verb is a reviewable governance change (OMN-14168).
``git`` and image-digest promotion are intentionally absent — a source
change produces a diff and therefore a PR, which must route through the
normal merged-PR gate. Imported lazily so this low-level model module
never triggers the heavy ``omnibase_core.validation`` package at import
time.
"""
if v is None:
return None
normalized = v.strip()
if not normalized:
raise ValueError("mutation_verb must contain non-whitespace characters")
# Local import avoids an import-time cycle through the validation package.
from omnibase_core.validation.runtime_ops_verb_loader import (
load_runtime_ops_verb_allowlist,
)

allowlist = load_runtime_ops_verb_allowlist()
if normalized not in allowlist:
raise ValueError(
f"mutation_verb {normalized!r} is not in the governed runtime-ops "
f"verb allowlist {sorted(allowlist)} "
"(contracts/runtime_ops_verb_allowlist.yaml). A source change "
"(git / image-digest promotion) produces a PR and must route "
"through the normal merged-PR gate, not the RUNTIME_OPS class."
)
return normalized

@model_validator(mode="after")
def enforce_adversarial_invariants(self) -> ModelDodReceipt:
"""Apply the four-rule Centralized Transition Policy.
Expand All @@ -399,7 +498,52 @@ def enforce_adversarial_invariants(self) -> ModelDodReceipt:

Rule (4) — schema_version SemVer match — is enforced as a field
validator above, before this method runs.

RUNTIME_OPS structural invariants (OMN-14168, G2a) are enforced first —
a receipt that declares ``evidence_class == RUNTIME_OPS`` must carry the
full runtime-ops mutation block, assert ``no_source_change``, and carry
NO ``pr_number`` (a PR routes through the normal merged-PR gate). These
raise ``ValueError`` because a mis-declared runtime-ops receipt is
structurally invalid, not merely weak.
"""
# RUNTIME_OPS structural invariants (G2a). A receipt of this class is a
# no-PR, no-source-change runtime-ops proof; it must be complete and must
# not smuggle a PR/source binding in.
if self.evidence_class is EnumEvidenceClass.RUNTIME_OPS:
if not self.no_source_change:
raise ValueError(
"evidence_class=RUNTIME_OPS requires no_source_change=True — "
"the runtime-ops class waives the merged-PR requirement only "
"for genuine no-source-change ops (OMN-14168, G2a)."
)
if self.pr_number is not None:
raise ValueError(
"evidence_class=RUNTIME_OPS must not carry a pr_number — work "
"with a PR routes through the normal merged-PR gate, not the "
"runtime-ops class (OMN-14168, G2a)."
)
if self.mutation_verb is None or not self.mutation_verb.strip():
raise ValueError(
"evidence_class=RUNTIME_OPS requires a mutation_verb from the "
"governed runtime-ops allowlist (OMN-14168)."
)
if self.mutation_command is None or not self.mutation_command.strip():
raise ValueError(
"evidence_class=RUNTIME_OPS requires a mutation_command (the "
"exact live-ops command applied) (OMN-14168)."
)
if self.target_identity is None or not self.target_identity.strip():
raise ValueError(
"evidence_class=RUNTIME_OPS requires a target_identity (the "
"mutated runtime object) (OMN-14168)."
)
if self.prevention_followup is None or not self.prevention_followup.strip():
raise ValueError(
"evidence_class=RUNTIME_OPS requires a prevention_followup "
"(a GitOps/declarative guardrail ticket or PR) so the no-PR-ops "
"hatch cannot become a permanent bypass (OMN-14168, G5)."
)

# Rule 3: executable check_types must have non-empty probe_stdout.
# PENDING is exempt — an unexecuted probe has no stdout by definition.
if (
Expand Down
82 changes: 82 additions & 0 deletions src/omnibase_core/validation/runtime_ops_verb_loader.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
# SPDX-License-Identifier: MIT

"""Runtime-ops verb allowlist loader (OMN-14168).

Resolves the bounded set of ``mutation_verb`` values a RUNTIME_OPS_READBACK
evidence receipt may carry. The allowlist is GOVERNED DATA — it lives in
``contracts/runtime_ops_verb_allowlist.yaml`` (mirroring
``test_selection_adjacency.yaml``), never hardcoded in Python — so adding or
removing a verb is a reviewable governance change, not a code edit.

Both enforcement surfaces resolve the set from here so they cannot drift:

* :class:`omnibase_core.models.contracts.ticket.model_dod_receipt.ModelDodReceipt`
validates ``mutation_verb`` against it, and
* the omnimarket ``DurableEvidenceGate`` RUNTIME_OPS_READBACK branch (G2b).

The result is cached per-process (``functools.cache``) so the bundled YAML is
read at most once; callers get a frozenset for cheap membership checks.
"""

from __future__ import annotations

import importlib.resources
from functools import cache
from pathlib import Path

import yaml

_CONTRACTS_PKG = "omnibase_core.contracts"
_ALLOWLIST_YAML = "runtime_ops_verb_allowlist.yaml"
_ALLOWLIST_KEY = "runtime_ops_verbs"


@cache
def load_runtime_ops_verb_allowlist() -> frozenset[str]:
"""Return the governed runtime-ops ``mutation_verb`` allowlist.

Reads ``contracts/runtime_ops_verb_allowlist.yaml`` via
``importlib.resources`` so it resolves both in a source checkout and in an
installed wheel. Result is cached for the process.

Raises:
FileNotFoundError: the bundled allowlist YAML is missing (fatal config
error — the class cannot be enforced without its governed data).
ValueError: the YAML exists but does not declare a non-empty
``runtime_ops_verbs`` list of strings.
"""
try:
ref = importlib.resources.files(_CONTRACTS_PKG) / _ALLOWLIST_YAML
raw = ref.read_text(encoding="utf-8")
except (FileNotFoundError, TypeError) as exc:
fallback = Path(__file__).parent.parent / "contracts" / _ALLOWLIST_YAML
if fallback.exists():
raw = fallback.read_text(encoding="utf-8")
else:
raise FileNotFoundError( # error-ok: bundled governed YAML missing — fatal config error
f"Cannot locate {_ALLOWLIST_YAML}; tried importlib.resources "
f"and {fallback}"
) from exc

data = yaml.safe_load(raw)
if not isinstance(data, dict):
raise ValueError(
f"{_ALLOWLIST_YAML} must be a mapping with a {_ALLOWLIST_KEY!r} key"
)
verbs = data.get(_ALLOWLIST_KEY)
if not isinstance(verbs, list) or not verbs:
raise ValueError(
f"{_ALLOWLIST_YAML} must declare a non-empty {_ALLOWLIST_KEY!r} list"
)
normalized: set[str] = set()
for verb in verbs:
if not isinstance(verb, str) or not verb.strip():
raise ValueError(
f"{_ALLOWLIST_KEY} entries must be non-blank strings, got: {verb!r}"
)
normalized.add(verb.strip())
return frozenset(normalized)


__all__ = ["load_runtime_ops_verb_allowlist"]
Loading
Loading