"""The skill bodies this worker was spawned with.
A worker never reads a skills root: the host validates the roots, decides what is
offered and what is refused, and hands the result over as a spawn payload. The
worker's job is only to make that payload importable, so an id a cell can reach is
an id the host offered, and the set stays fixed for the life of the generation.
A skill body is a trail, not a specification: it is the sequence that leads to the
state the id names. A description stored beside it would be a second, staler
definition of the same thing, which is why there is no such field to read here.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import json
@dataclass(frozen=True, slots=True)
class Body:
"""One offered skill, with the digest the host computed for its bytes."""
id: str
origin: str
path: str
digest: str
text: str
@dataclass(frozen=True, slots=True)
class Refusal:
"""An id that exists on disk and was not offered, with the reason."""
id: str
reason: str
_loaded = False
_revision: str | None = None
_roots: dict[str, str | None] = {}
_bodies: tuple[Body, ...] = ()
_refused: tuple[Refusal, ...] = ()
def install(path: str | Path) -> None:
"""Read the spawn payload. Absence means no skills, never an error.
Installing replaces the whole set: a payload that cannot be read or parsed leaves
this process with no skills, never with the previous set, because a stale body is
exactly the half-read state a later reader would take for the truth.
"""
global _loaded, _revision, _roots, _bodies, _refused
_loaded = False
_revision = None
_roots = {}
_bodies = ()
_refused = ()
try:
raw: Any = json.loads(Path(path).read_text(encoding="utf-8"))
except (OSError, ValueError):
return
if not isinstance(raw, dict):
return
_loaded = True
_revision = raw.get("revision") if isinstance(raw.get("revision"), str) else None
reported = raw.get("roots")
if isinstance(reported, dict):
_roots = {
str(name): value if isinstance(value, str) else None
for name, value in reported.items()
}
_bodies = tuple(
Body(
id=str(entry["id"]),
origin=str(entry.get("origin", "builtin")),
path=str(entry.get("path", "")),
digest=str(entry.get("digest", "")),
text=str(entry["text"]),
)
for entry in raw.get("skills", [])
if isinstance(entry, dict) and isinstance(entry.get("id"), str) and "text" in entry
)
_refused = tuple(
Refusal(id=str(entry.get("id", "")), reason=str(entry.get("reason", "")))
for entry in raw.get("refused", [])
if isinstance(entry, dict)
)
def loaded() -> bool:
"""Whether a payload was read. False means this process has no skill set."""
return _loaded
def roots() -> dict[str, str | None]:
"""Where the offered ids were read from, as the host reported them.
Present so that an id missing from this generation is a statement about the scan
that ran, not a hole a reader has to explain by inspecting a filesystem.
"""
return dict(_roots)
def revision() -> str:
"""Content identity of the whole offered set, or a named absence.
Reported as a value rather than `None` so a caller that prints it cannot
mistake an empty set for a missing one.
"""
if _revision is None:
return "no-skills"
return _revision
def bodies() -> tuple[Body, ...]:
return _bodies
def refused() -> tuple[Refusal, ...]:
return _refused
__all__ = [
"Body",
"Refusal",
"bodies",
"install",
"loaded",
"refused",
"revision",
"roots",
]