| 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) |
|
|
