diff --git a/src/omnimarket/config/state_root.py b/src/omnimarket/config/state_root.py new file mode 100644 index 0000000000..87319fa060 --- /dev/null +++ b/src/omnimarket/config/state_root.py @@ -0,0 +1,65 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""The runtime's writable state root, resolved fail-fast from its declared env. + +OMN-19654. A deployed runtime pod runs with ``readOnlyRootFilesystem: true`` +and ``HOME=/``, so any state written under a ``Path.home()``-derived default +lands on ``/`` and fails with ``[Errno 30] Read-only file system``. The runtime +manifests already declare where writable state goes -- ``ONEX_STATE_DIR`` +(the onex-dev family, including the onex-lab lane) and ``ONEX_STATE_ROOT`` +(both that family and the prod runtime ConfigMap), each backed by a writable +volume -- so state that belongs to a running runtime is resolved from those, +in that order, and nowhere else. + +There is no fallback. When neither is set this raises a named error rather +than choosing a home, a working directory or a temp directory: a silent +default is how the claim store ended up on a read-only root in the first place +(CLAUDE.md rules 6 and 8). + +These two keys are bootstrap-only by the OMN-10548 settings design, which is +why this reads them directly rather than through ``Settings``; the read is +confined to this one module so that callers do not each re-derive it. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Final + +ONEX_STATE_ROOT_ENV_KEYS: Final[tuple[str, ...]] = ("ONEX_STATE_DIR", "ONEX_STATE_ROOT") + + +class OnexStateRootUnconfiguredError(RuntimeError): + """No writable runtime state root is declared for this process.""" + + def __init__(self, purpose: str) -> None: + super().__init__( + f"{purpose} needs a writable runtime state root, and none is " + f"configured: set one of {', '.join(ONEX_STATE_ROOT_ENV_KEYS)} to a " + "writable directory (a deployed runtime mounts a volume there). " + "There is no home-directory or working-directory fallback " + "(OMN-19654)." + ) + self.purpose = purpose + + +def resolve_onex_state_root(*, purpose: str) -> Path: + """Return the declared state root, or raise naming ``purpose`` and the keys. + + ``ONEX_STATE_DIR`` wins over ``ONEX_STATE_ROOT``, matching every other + runtime state-root reader in this package. A value that is empty or only + whitespace counts as unset. + """ + for key in ONEX_STATE_ROOT_ENV_KEYS: + value = os.environ.get(key, "").strip() + if value: + return Path(value) + raise OnexStateRootUnconfiguredError(purpose) + + +__all__ = [ + "ONEX_STATE_ROOT_ENV_KEYS", + "OnexStateRootUnconfiguredError", + "resolve_onex_state_root", +] diff --git a/src/omnimarket/nodes/node_delegate_skill_orchestrator/ports/port_delegation_claim.py b/src/omnimarket/nodes/node_delegate_skill_orchestrator/ports/port_delegation_claim.py index 8b352224ee..acdcd0e900 100644 --- a/src/omnimarket/nodes/node_delegate_skill_orchestrator/ports/port_delegation_claim.py +++ b/src/omnimarket/nodes/node_delegate_skill_orchestrator/ports/port_delegation_claim.py @@ -47,6 +47,8 @@ from __future__ import annotations import json +import threading +from collections.abc import Callable from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path @@ -113,6 +115,26 @@ def record_terminal( ... +def _require_attested_write( + database: ProtocolProjectionDatabaseSync, +) -> ProtocolProjectionAttestedWrite: + # The claim needs `upsert_returning`, which lives on the NARROWER + # attested-write protocol rather than on the sync protocol every + # adapter implements. That separation is deliberate upstream, and the + # documented contract for a caller that needs it is to probe and + # refuse LOUDLY naming the adapter, rather than to assume. A claim + # that silently could not be made would let the double-bill straight + # through, which is the one outcome this port exists to prevent. + if not isinstance(database, ProtocolProjectionAttestedWrite): + raise TypeError( + f"{type(database).__name__} does not implement " + "ProtocolProjectionAttestedWrite, so a delegate-skill command " + "claim cannot be made atomically and a redelivery would " + "re-run and re-bill the inference (OMN-18887)" + ) + return database + + class DelegationClaimPort: """``ProtocolDelegationIdempotencyPort`` over the sync projection adapter. @@ -121,28 +143,52 @@ class DelegationClaimPort: selects, so AC4's durability does not depend on which one is configured. """ - def __init__(self, database: ProtocolProjectionDatabaseSync) -> None: - # The claim needs `upsert_returning`, which lives on the NARROWER - # attested-write protocol rather than on the sync protocol every - # adapter implements. That separation is deliberate upstream, and the - # documented contract for a caller that needs it is to probe and - # refuse LOUDLY naming the adapter, rather than to assume. A claim - # that silently could not be made would let the double-bill straight - # through, which is the one outcome this port exists to prevent. - if not isinstance(database, ProtocolProjectionAttestedWrite): + def __init__( + self, + database: ProtocolProjectionDatabaseSync | None = None, + *, + resolve_database: Callable[[], ProtocolProjectionDatabaseSync] | None = None, + ) -> None: + # OMN-19654: `resolve_database` defers choosing the backing store to + # the first claim. A claim is only ever attempted for a bus delivery, + # so the bus-less CLI -- which builds this port through the same + # handler and never claims -- does not need a runtime state root to + # exist, while a runtime that does claim is still refused, loudly, at + # the claim, when it declares none. Exactly one of the two is given. + if (database is None) == (resolve_database is None): raise TypeError( - f"{type(database).__name__} does not implement " - "ProtocolProjectionAttestedWrite, so a delegate-skill command " - "claim cannot be made atomically and a redelivery would " - "re-run and re-bill the inference (OMN-18887)" + "DelegationClaimPort takes exactly one of `database` or " + "`resolve_database`" ) - self._database: ProtocolProjectionAttestedWrite = database + self._resolve_database = resolve_database + self._resolved: ProtocolProjectionAttestedWrite | None = ( + _require_attested_write(database) if database is not None else None + ) + # Four records run in flight in one process (OMN-18852), so the first + # claims on a fresh port can race. The deferred store is resolved + # under this lock so exactly one adapter is built and every caller + # claims through it. + self._resolve_lock = threading.Lock() + + def _database(self) -> ProtocolProjectionAttestedWrite: + resolved = self._resolved + if resolved is not None: + return resolved + with self._resolve_lock: + if self._resolved is None: + if self._resolve_database is None: + raise RuntimeError( + "DelegationClaimPort has neither a database nor a " + "resolver (OMN-19654)" + ) + self._resolved = _require_attested_write(self._resolve_database()) + return self._resolved def claim( self, *, delivery_id: UUID, correlation_id: UUID ) -> ModelDelegationClaimOutcome: mine = datetime.now(UTC).isoformat() - rows = self._database.upsert_returning( + rows = self._database().upsert_returning( CLAIMS_TABLE, _DELIVERY_COLUMN, { @@ -193,7 +239,7 @@ def record_terminal( # rather than falling through to the update. Passing it insert-only # satisfies the INSERT arm while leaving the real claimer's timestamp # untouched on the arm that actually runs. - self._database.upsert_returning( + self._database().upsert_returning( CLAIMS_TABLE, _DELIVERY_COLUMN, { @@ -206,16 +252,28 @@ def record_terminal( def default_claim_db_path() -> Path: - """The local claim store, beside the evidence store but NOT the same file. - - Control state and evidence are different things and are kept in different - places on purpose. Sharing the evidence file would also couple this - store's lifecycle to the evidence target's, so clearing one to reset a - local install would silently reopen the double-bill. + """The local claim store, under the runtime's declared state root. + + OMN-19654: this used to sit beside the evidence store, whose location is + derived from ``Path.home()``. A deployed runtime pod has ``HOME=/`` on a + read-only root filesystem, so on the onex-lab lane every bus delivery + tried to create ``/.omninode`` and terminalized ``OSError: [Errno 30] + Read-only file system``. A claim is only made for a bus delivery, i.e. by + a running runtime, and a running runtime declares where its writable + state lives, so the store is resolved from that declaration and raises + :class:`~omnimarket.config.state_root.OnexStateRootUnconfiguredError` + when there is none. + + It is still NOT the evidence file. Control state and evidence are + different things and are kept in different places on purpose: sharing + the evidence file would couple this store's lifecycle to the evidence + target's, so clearing one to reset a local install would silently reopen + the double-bill. """ - from omnimarket.projection.sqlite_database import default_evidence_db_path + from omnimarket.config.state_root import resolve_onex_state_root - return default_evidence_db_path().with_name("delegation_claims.sqlite") + state_root = resolve_onex_state_root(purpose="the delegate-skill claim store") + return state_root / "delegation" / "delegation_claims.sqlite" def resolve_delegation_claim_store() -> DelegationClaimPort: @@ -223,7 +281,8 @@ def resolve_delegation_claim_store() -> DelegationClaimPort: Follows the projection binding overlay exactly as the evidence target does, so a deployment pointing delegation at Postgres gets its claims - there too, and falls back to a local SQLite file of its own otherwise. + there too, and falls back to a local SQLite file of its own otherwise, + under the runtime's declared state root (OMN-19654). """ from omnimarket.nodes.node_delegate_skill_orchestrator.ports.evidence_db_resolution import ( _adapter_for_dsn, @@ -235,7 +294,11 @@ def resolve_delegation_claim_store() -> DelegationClaimPort: binding = projection_runtime_binding_from_overlay_env() if binding is None: - return DelegationClaimPort(SqliteDatabaseAdapter(default_claim_db_path())) + # Deferred, so the state root is resolved (and refused if absent) at + # the first claim rather than when the handler is built (OMN-19654). + return DelegationClaimPort( + resolve_database=lambda: SqliteDatabaseAdapter(default_claim_db_path()) + ) return DelegationClaimPort(_adapter_for_dsn(binding.resolve_database_url())) diff --git a/tests/unit/nodes/node_delegate_skill_orchestrator/test_claim_store_state_root_omn19654.py b/tests/unit/nodes/node_delegate_skill_orchestrator/test_claim_store_state_root_omn19654.py new file mode 100644 index 0000000000..eec946e3df --- /dev/null +++ b/tests/unit/nodes/node_delegate_skill_orchestrator/test_claim_store_state_root_omn19654.py @@ -0,0 +1,234 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""OMN-19654: the delegate-skill claim store lives under the configured state root. + +On the onex-lab k3s lane the deployed ``node_delegate_skill_orchestrator`` ran +in a pod with ``readOnlyRootFilesystem: true`` and ``HOME=/``. With no +projection binding overlay on that pod, the OMN-18887 claim port fell back to a +local SQLite file whose location was derived from ``Path.home()``, so every +bus delivery tried to create ``/.omninode`` and terminalized +``OSError: [Errno 30] Read-only file system: '/.omninode'``. No delegation +completed on the lane. + +These tests pin the fix: the fallback claim store is resolved from the +runtime's declared state root (``ONEX_STATE_DIR``, then ``ONEX_STATE_ROOT``), +never from the home directory, and an unconfigured state root is refused with a +named error at the moment a claim is attempted -- not silently defaulted, and +not at handler construction, where the bus-less CLI (which never claims) would +be broken for nothing. + +The session conftest replaces ``default_claim_db_path`` on the claim module to +isolate the suite's claims. These tests need the REAL resolution, so they load +a fresh, unpatched copy of the module from its source file. +""" + +from __future__ import annotations + +import importlib.util +import stat +import sys +from collections.abc import Iterator +from pathlib import Path +from types import ModuleType +from uuid import uuid4 + +import pytest + +from omnimarket.nodes.node_delegate_skill_orchestrator.ports import ( + port_delegation_claim as _patched_claim_module, +) + +_STATE_ENV_KEYS = ("ONEX_STATE_DIR", "ONEX_STATE_ROOT") +_BINDING_OVERLAY_ENV = "OMNIMARKET_PROJECTION_RUNTIME_BINDING_OVERLAY" +_CLAIM_DB_RELATIVE = Path("delegation") / "delegation_claims.sqlite" + + +@pytest.fixture +def fresh_claim_module() -> Iterator[ModuleType]: + """An unpatched instance of the claim module, loaded from its source file.""" + source = _patched_claim_module.__file__ + assert source is not None + name = "_omn19654_fresh_port_delegation_claim" + spec = importlib.util.spec_from_file_location(name, source) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + try: + spec.loader.exec_module(module) + yield module + finally: + sys.modules.pop(name, None) + + +@pytest.fixture +def readonly_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Path]: + """A HOME that cannot be written, standing in for ``HOME=/`` on a ro rootfs.""" + home = tmp_path / "readonly-home" + home.mkdir() + home.chmod(stat.S_IRUSR | stat.S_IXUSR) + monkeypatch.setenv("HOME", str(home)) + # The evidence path is an import-time constant built from the HOME the + # process started with. Point it into the read-only home too, so a + # regression to a home-derived claim path fails here instead of writing + # into the real home directory of whoever runs the suite. + from omnimarket.projection import sqlite_database + + monkeypatch.setattr( + sqlite_database, + "_DEFAULT_EVIDENCE_DB_PATH", + home / ".omninode" / "delegation" / "delegation.sqlite", + ) + monkeypatch.delenv(_BINDING_OVERLAY_ENV, raising=False) + for key in _STATE_ENV_KEYS: + monkeypatch.delenv(key, raising=False) + try: + yield home + finally: + home.chmod(stat.S_IRWXU) + + +@pytest.mark.unit +def test_omninode_readonly_rootfs_claim_lands_under_onex_state_dir( + fresh_claim_module: ModuleType, + readonly_home: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state_dir = tmp_path / "state-dir" + monkeypatch.setenv("ONEX_STATE_DIR", str(state_dir)) + + port = fresh_claim_module.resolve_delegation_claim_store() + outcome = port.claim(delivery_id=uuid4(), correlation_id=uuid4()) + + assert outcome.won is True + assert (state_dir / _CLAIM_DB_RELATIVE).is_file() + assert not (readonly_home / ".omninode").exists() + + +@pytest.mark.unit +def test_omninode_readonly_rootfs_claim_uses_onex_state_root_when_dir_unset( + fresh_claim_module: ModuleType, + readonly_home: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The prod runtime ConfigMap declares ONEX_STATE_ROOT and not ONEX_STATE_DIR. + state_root = tmp_path / "state-root" + monkeypatch.setenv("ONEX_STATE_ROOT", str(state_root)) + + assert fresh_claim_module.default_claim_db_path() == ( + state_root / _CLAIM_DB_RELATIVE + ) + port = fresh_claim_module.resolve_delegation_claim_store() + assert port.claim(delivery_id=uuid4(), correlation_id=uuid4()).won is True + assert not (readonly_home / ".omninode").exists() + + +@pytest.mark.unit +def test_omninode_readonly_rootfs_onex_state_dir_takes_precedence( + fresh_claim_module: ModuleType, + readonly_home: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ONEX_STATE_DIR", str(tmp_path / "dir")) + monkeypatch.setenv("ONEX_STATE_ROOT", str(tmp_path / "root")) + + assert fresh_claim_module.default_claim_db_path() == ( + tmp_path / "dir" / _CLAIM_DB_RELATIVE + ) + + +@pytest.mark.unit +@pytest.mark.parametrize("blank", ["", " "]) +def test_omninode_readonly_rootfs_unconfigured_state_root_refuses_at_claim( + fresh_claim_module: ModuleType, + readonly_home: Path, + monkeypatch: pytest.MonkeyPatch, + blank: str, +) -> None: + for key in _STATE_ENV_KEYS: + monkeypatch.setenv(key, blank) + + # Construction must not fail: the bus-less CLI builds this port and never + # claims, because a claim is only attempted on a bus delivery. + port = fresh_claim_module.resolve_delegation_claim_store() + + from omnimarket.config.state_root import OnexStateRootUnconfiguredError + + with pytest.raises(OnexStateRootUnconfiguredError) as excinfo: + port.claim(delivery_id=uuid4(), correlation_id=uuid4()) + + message = str(excinfo.value) + for key in _STATE_ENV_KEYS: + assert key in message + assert not (readonly_home / ".omninode").exists() + + +@pytest.mark.unit +def test_omninode_readonly_rootfs_claim_store_is_durable_across_ports( + fresh_claim_module: ModuleType, + readonly_home: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # A redelivery handled by a second port instance (a restarted consumer on + # the same volume) must lose the claim the first instance won. + monkeypatch.setenv("ONEX_STATE_DIR", str(tmp_path / "state")) + delivery_id = uuid4() + first = fresh_claim_module.resolve_delegation_claim_store() + second = fresh_claim_module.resolve_delegation_claim_store() + + assert first.claim(delivery_id=delivery_id, correlation_id=uuid4()).won is True + assert second.claim(delivery_id=delivery_id, correlation_id=uuid4()).won is False + + +@pytest.mark.unit +def test_omninode_readonly_rootfs_concurrent_first_claims_resolve_store_once( + fresh_claim_module: ModuleType, + readonly_home: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Four records run in flight in one process (OMN-18852), so the first + # claims on a fresh port can arrive together. The deferred store must be + # resolved exactly once, and exactly one of the concurrent deliveries of + # the same record may win. + import threading + import time + + from omnimarket.projection.sqlite_database import SqliteDatabaseAdapter + + monkeypatch.setenv("ONEX_STATE_DIR", str(tmp_path / "state")) + workers = 8 + resolutions: list[int] = [] + resolutions_lock = threading.Lock() + + def slow_resolve() -> SqliteDatabaseAdapter: + with resolutions_lock: + resolutions.append(1) + time.sleep(0.05) + return SqliteDatabaseAdapter(fresh_claim_module.default_claim_db_path()) + + port = fresh_claim_module.DelegationClaimPort(resolve_database=slow_resolve) + delivery_id = uuid4() + barrier = threading.Barrier(workers) + wins: list[bool] = [] + wins_lock = threading.Lock() + + def claim_once() -> None: + barrier.wait() + outcome = port.claim(delivery_id=delivery_id, correlation_id=uuid4()) + with wins_lock: + wins.append(outcome.won) + + threads = [threading.Thread(target=claim_once) for _ in range(workers)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert len(resolutions) == 1 + assert len(wins) == workers + assert wins.count(True) == 1