Something went wrong. Try again.
ive harnessed the harness
Something went wrong. Try again.
2.0 kB · 45 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546"""One persistent policy process; no scheduler or agent loop."""from __future__ import annotations
import importlibimport inspectimport sysfrom pathlib import Path
# Load the SDK before adding editable behavior to the import path.from klbr.hooks import (AttentionDecision, AttentionRequest, DeliveryDecision, DeliveryRequest, HookContext)from .release import fingerprint, manifest
class Policy: def __init__(self, root: Path, revision: str): if fingerprint(root) != revision: raise ValueError("behavior content does not match its revision fingerprint") registrations = manifest(root) sys.path.insert(0, str(root.resolve())) self.revision = revision self.handlers = {} for slot, entrypoint in registrations.items(): module, name = entrypoint.split(":") handler = getattr(importlib.import_module(module), name) if not callable(handler): raise ValueError(f"not a callable: {entrypoint}") # Enforce the two-argument use site without evaluating the callback. inspect.signature(handler).bind(object(), object()) self.handlers[slot] = handler
async def invoke(self, slot: str, value: dict, context: HookContext) -> dict: match slot: case "delivery.review": request, expected = DeliveryRequest.parse(value), DeliveryDecision case "attention.plan": request, expected = AttentionRequest.parse(value), AttentionDecision case _: raise ValueError(f"unsupported hook slot: {slot}") result = self.handlers[slot](request, context) if inspect.isawaitable(result): result = await result if not isinstance(result, expected): raise TypeError(f"{slot} must return {expected.__name__}, not {type(result).__name__}") return result.to_wire()