"""Content identity shared with Rust's behavior::Release.
This module verifies a supplied immutable snapshot. It does not choose the active
revision or mutate a host database. Filesystem permissions are not a sandbox.
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
MAX_FILES = 128
MAX_BYTES = 2_097_152
SLOTS = {"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