108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""
|
|
Structured Rejection Envelope — {received, reason, allowed, hint}.
|
|
|
|
Every blocked action returns a standard 4-field envelope so the verifier is
|
|
self-documenting: the error message IS the documentation.
|
|
|
|
Inspired by pi-agents's blocked-action format, ported to be Agno-agnostic.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Optional
|
|
|
|
|
|
@dataclass
|
|
class RejectionEnvelope:
|
|
"""Universal envelope for blocked actions.
|
|
|
|
Fields:
|
|
received: What the agent tried to do (verbatim action or intent)
|
|
reason: Why it was blocked (machine-readable short code + human explanation)
|
|
allowed: What the agent CAN do instead (alternative action, or None if completely denied)
|
|
hint: How to resolve the rejection (docs reference, fix instruction, or None)
|
|
"""
|
|
|
|
received: str
|
|
reason: str
|
|
allowed: Optional[str] = None
|
|
hint: Optional[str] = None
|
|
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
|
|
|
|
def dict(self) -> dict[str, Any]:
|
|
return {
|
|
"received": self.received,
|
|
"reason": self.reason,
|
|
"allowed": self.allowed,
|
|
"hint": self.hint,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
|
|
def to_json(self, indent: int = 2) -> str:
|
|
return json.dumps(self.dict(), indent=indent)
|
|
|
|
def __str__(self) -> str:
|
|
parts = [f"BLOCKED: {self.reason}"]
|
|
if self.allowed:
|
|
parts.append(f" Allowed: {self.allowed}")
|
|
if self.hint:
|
|
parts.append(f" Hint: {self.hint}")
|
|
return "\n".join(parts)
|
|
|
|
@classmethod
|
|
def from_json(cls, raw: str) -> RejectionEnvelope:
|
|
data = json.loads(raw)
|
|
return cls(
|
|
received=data["received"],
|
|
reason=data["reason"],
|
|
allowed=data.get("allowed"),
|
|
hint=data.get("hint"),
|
|
timestamp=data.get("timestamp", ""),
|
|
)
|
|
|
|
@classmethod
|
|
def block(
|
|
cls,
|
|
received: str,
|
|
reason: str,
|
|
allowed: Optional[str] = None,
|
|
hint: Optional[str] = None,
|
|
) -> RejectionEnvelope:
|
|
"""Convenience constructor for a blocking rejection."""
|
|
return cls(received=received, reason=reason, allowed=allowed, hint=hint)
|
|
|
|
|
|
# ── Blocker: a rule that generates RejectionEnvelopes ──────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class Blocker:
|
|
"""A named rule that evaluates an action and returns a rejection or None.
|
|
|
|
Usage::
|
|
|
|
no_write_file = Blocker(
|
|
name="no_direct_write",
|
|
match=lambda action: "write_file" in action.get("tool", ""),
|
|
reject=lambda action: RejectionEnvelope.block(
|
|
received=str(action),
|
|
reason="LEADER_MUST_NOT_WRITE_FILES",
|
|
allowed="Use delegate_task to spawn a writer subagent",
|
|
hint="Leaders read, delegate, and synthesize — they never write files directly.",
|
|
),
|
|
)
|
|
"""
|
|
|
|
name: str
|
|
match: Any # Callable[[dict], bool]
|
|
reject: Any # Callable[[dict], RejectionEnvelope]
|
|
|
|
def evaluate(self, action: dict) -> Optional[RejectionEnvelope]:
|
|
"""Return a RejectionEnvelope if this action is blocked, None otherwise."""
|
|
if self.match(action):
|
|
return self.reject(action)
|
|
return None
|