File size: 2,143 Bytes
945aae0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | import logging
from typing import Callable
logger = logging.getLogger("qkernel.policy")
class PolicyContext:
def __init__(self):
self.allowed: bool = True
self.route_override: str | None = None
self.priority_boost: int = 0
self.tags: list[str] = []
class PolicyRule:
def __init__(
self,
name: str,
condition: Callable[[dict], bool],
action: Callable[[dict, PolicyContext], PolicyContext],
):
self.name = name
self.condition = condition
self.action = action
class PolicyEngine:
def __init__(self):
self._rules: list[PolicyRule] = []
def add_rule(self, rule: PolicyRule) -> None:
self._rules.append(rule)
def evaluate(self, event: dict) -> PolicyContext:
ctx = PolicyContext()
for rule in self._rules:
if rule.condition(event):
logger.debug("Policy rule '%s' triggered", rule.name)
ctx = rule.action(event, ctx)
return ctx
def add_default_rules(self) -> None:
self.add_rule(PolicyRule(
name="block_anonymous",
condition=lambda e: e.get("user_id", "anonymous") == "anonymous",
action=lambda e, ctx: setattr(ctx, "allowed", False) or ctx,
))
block_anon_rule = PolicyRule(
name="block_anonymous_message",
condition=lambda e: e.get("user_id", "anonymous") == "anonymous"
and bool(e.get("message", "").strip()),
action=lambda e, ctx: setattr(ctx, "allowed", False) or ctx,
)
self.add_rule(block_anon_rule)
chat_boost = PolicyRule(
name="boost_chat",
condition=lambda e: e.get("type") == "chat",
action=lambda e, ctx: setattr(ctx, "priority_boost", 5) or ctx,
)
self.add_rule(chat_boost)
long_message = PolicyRule(
name="long_message_route",
condition=lambda e: len(e.get("message", "")) > 500,
action=lambda e, ctx: setattr(ctx, "route_override", "brain") or ctx,
)
self.add_rule(long_message)
|