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
65 changes: 65 additions & 0 deletions src/omnimarket/config/state_root.py
Original file line number Diff line number Diff line change
@@ -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",
]
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.

Expand All @@ -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,
{
Expand Down Expand Up @@ -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,
{
Expand All @@ -206,24 +252,37 @@ 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:
"""Resolve the claim port against the configured substrate.

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,
Expand All @@ -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()))


Expand Down
Loading
Loading