ive harnessed the harness
Something went wrong. Try again.
3.8 kB · 134 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135"""The skill bodies this worker was spawned with.
A worker never reads a skills root: the host validates the roots, decides what isoffered and what is refused, and hands the result over as a spawn payload. Theworker's job is only to make that payload importable, so an id a cell can reach isan 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 thestate the id names. A description stored beside it would be a second, stalerdefinition of the same thing, which is why there is no such field to read here."""from __future__ import annotations
from dataclasses import dataclassfrom pathlib import Pathfrom 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",]