ive harnessed the harness
Something went wrong. Try again.
6.1 kB · 173 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174"""Content identity shared with Rust's behavior::Release.
This module verifies a supplied immutable snapshot. It does not choose the activerevision or mutate a host database. Filesystem permissions are not a sandbox."""from __future__ import annotations
import hashlibimport jsonfrom pathlib import Path
MAX_FILES = 128MAX_BYTES = 2_097_152SLOTS = {"attention.plan", "delivery.review"}
def files(root: Path) -> list[tuple[str, bytes]]: if root.is_symlink() or not root.is_dir(): raise ValueError("behavior root must be a real directory") result = [] total = 0 for path in sorted(root.rglob("*")): if path.is_symlink(): raise ValueError("behavior snapshots cannot contain symlinks") if path.is_dir(): continue if path.suffix not in {".py", ".json", ".md"}: raise ValueError(f"unsupported behavior file: {path.name}") if not path.is_file() or path.stat().st_size > MAX_BYTES: raise ValueError("behavior file is not regular or exceeds limit") # Read one byte beyond the remaining bound to catch concurrent growth # without allocating an arbitrarily large draft file. with path.open("rb") as stream: content = stream.read(MAX_BYTES - total + 1) total += len(content) if total > MAX_BYTES or len(result) >= MAX_FILES: raise ValueError("behavior snapshot exceeds limits") result.append((path.relative_to(root).as_posix(), content)) return sorted(result)
def fingerprint(root: Path) -> str: digest = hashlib.sha256() for name, content in files(root): name_bytes = name.encode("utf-8") digest.update(len(name_bytes).to_bytes(8, "big")) digest.update(name_bytes) digest.update(len(content).to_bytes(8, "big")) digest.update(content) return digest.hexdigest()
def manifest(root: Path) -> dict[str, str]: from .wire import fields value = fields(json.loads((root / "manifest.json").read_text()), {"version", "hooks"}) if type(value["version"]) is not int or value["version"] != 1: raise ValueError("unsupported behavior manifest") hooks = value["hooks"] if not isinstance(hooks, dict) or set(hooks) != SLOTS: raise ValueError(f"this slice requires exactly {sorted(SLOTS)}") for entrypoint in hooks.values(): if not isinstance(entrypoint, str) or entrypoint.count(":") != 1: raise ValueError("expected module:function entrypoint") module, function = entrypoint.split(":") if not module.startswith("klbr_hooks.") or not all(x.isidentifier() for x in module.split(".")) or not function.isidentifier(): raise ValueError("hook entrypoints must be functions in klbr_hooks modules") return hooks
from dataclasses import dataclass, field
@dataclass(frozen=True, slots=True)class DeploymentSeed: seed_id: str revision: str source: str created_at_ms: int = 0
@dataclass(frozen=True, slots=True)class ReconciliationConflict: base_seed: str live_selection: str new_seed: str message: str = field(default="")
def __post_init__(self): if not self.message: object.__setattr__( self, "message", f"three-way reconciliation conflict: base seed '{self.base_seed}', live selection '{self.live_selection}', new seed '{self.new_seed}'", )
@dataclass(frozen=True, slots=True)class ReconciliationOutcome: action: str # "preserved", "fast_forward", "converged", "conflict" effective_revision: str conflict: ReconciliationConflict | None = None
@property def is_conflict(self) -> bool: return self.action == "conflict"
def reconcile_seed( base_seed: str, live_selection: str, new_seed: str,) -> ReconciliationOutcome: """Three-way reconciliation for deployment seeds vs live selection (P13, CONTRACTS Section 10).
Rules: 1. If new_seed == base_seed: deployment seed did not change. Live selection preserved. 2. If new_seed != base_seed and live_selection == base_seed: no live divergence, fast-forward to new seed. 3. If new_seed == live_selection: converged to same revision. 4. If new_seed != base_seed and live_selection != base_seed and new_seed != live_selection: Conflict! Reports conflict without silently overwriting either side. """ if new_seed == base_seed: return ReconciliationOutcome( action="preserved", effective_revision=live_selection, ) if live_selection == base_seed: return ReconciliationOutcome( action="fast_forward", effective_revision=new_seed, ) if new_seed == live_selection: return ReconciliationOutcome( action="converged", effective_revision=live_selection, ) # Both diverged and disagree: report conflict without overwriting either side conflict = ReconciliationConflict( base_seed=base_seed, live_selection=live_selection, new_seed=new_seed, ) return ReconciliationOutcome( action="conflict", effective_revision=live_selection, conflict=conflict, )
@dataclass(slots=True)class RevisionStatus: selected_revision: str applied_revision: str last_good_revision: str
def __init__(self, initial: str) -> None: self.selected_revision = initial self.applied_revision = initial self.last_good_revision = initial
def select(self, new_revision: str) -> None: self.selected_revision = new_revision
def apply_success(self, applied: str) -> None: self.applied_revision = applied self.last_good_revision = applied
def apply_failure(self, failed_revision: str) -> None: # Broken candidate start preserves the last good worker/revision! self.applied_revision = self.last_good_revision
@property def in_sync(self) -> bool: return self.selected_revision == self.applied_revision