"""Who may talk to whom: the rules behind the crowd, `ask_friend` and the roster. Always evaluated **from the chat's main model**. If the main model may not talk to a target, the target is not *offered* -- not listed on its roster, not named as a friend it may ask, not in the crowd picker's main list. Members of a crowd are not checked against each other: the rule is about who a conversation's own model brings in, and a member loses `friend` and `subagent` anyway. Two answers, because the owner asked for two things: * **offered** -- what happens on its own: the roster, a friend a model names, the picker's main list. * **addable** -- what a person may do by hand in the crowd picker. A rule a person wrote for themselves is soft for them; an instance rule is hard, unless they hold `rules.override`. `decide` is pure -- plain values in, a `Verdict` out -- so every combination is tested without a database, and the admin page's matrix is drawn by the same function that enforces the rules, which is what makes the matrix trustworthy. **A different data group is an implicit deny.** A crowd member or a friend is sent the conversation, so a model in another group joins only when a rule says so explicitly: the administrator's, or a person's own when they hold the override. It then reads its own group's stores, never the chat's. """ from __future__ import annotations from dataclasses import dataclass from sqlalchemy import select from sqlalchemy.orm import Session as DBSession from lembas.db.models import ( ANY_MODEL, EFFECT_ALLOW, EFFECT_DENY, EFFECTS, Model, TalkRule, User, ) MODE_OPEN = "open" MODE_CLOSED = "closed" MODES = (MODE_OPEN, MODE_CLOSED) # Where a person's own mode is kept in `settings_json`. Empty follows the instance. SETTING_KEY = "talk_mode" # Lets a person's own rules and mode win over the instance's, for them alone. PERMISSION = "rules.override" @dataclass(frozen=True) class Rule: from_model: str to_model: str effect: str # Why something is not offered. A code rather than a sentence, so a screen can # say it in the reader's language (`api/admin_rules.py:describe`) while a model # refused a friend is told it in English (`Verdict.reason`). WHY_INSTANCE_RULE = "instance_rule" WHY_YOUR_RULE = "your_rule" WHY_GROUP = "group" WHY_INSTANCE_CLOSED = "instance_closed" WHY_YOUR_CLOSED = "your_closed" @dataclass(frozen=True) class Verdict: offered: bool addable: bool why: str = "" rule: Rule | None = None @property def reason(self) -> str: """The reason in English, for a model -- or "" when it is offered.""" if self.offered or not self.why: return "" if self.why in (WHY_INSTANCE_RULE, WHY_YOUR_RULE) and self.rule is not None: whose = "the instance's" if self.why == WHY_INSTANCE_RULE else "your" return _named(self.rule, whose) return { WHY_GROUP: "it is in another data group", WHY_INSTANCE_CLOSED: "the instance allows no model to talk to another", WHY_YOUR_CLOSED: "your setting allows no model to talk to another", }.get(self.why, "") def match(rules: list[Rule], main: str, target: str) -> Rule | None: """The most specific rule for a pair: exact, then `main -> *`, `* -> target`, `* -> *`.""" for wanted in ((main, target), (main, ANY_MODEL), (ANY_MODEL, target), (ANY_MODEL, ANY_MODEL)): for rule in rules: if (rule.from_model, rule.to_model) == wanted: return rule return None def _named(rule: Rule, whose: str) -> str: frm = "any model" if rule.from_model == ANY_MODEL else rule.from_model to = "any model" if rule.to_model == ANY_MODEL else rule.to_model verb = "allows" if rule.effect == EFFECT_ALLOW else "forbids" return f"{whose} rule {frm} → {to} {verb} it" def decide( *, instance_mode: str, instance_rule: Rule | None, user_mode: str = "", user_rule: Rule | None = None, override: bool = False, same_group: bool = True, ) -> Verdict: """Whether a main model may talk to a target, offered and by hand. The instance's verdict is its most specific rule, or failing that its mode -- with a different data group counting as a deny that only an explicit allow opens. Without the override a person can only narrow: their own rule or their `closed` mode can take something off what is offered, and since those are theirs, they may still add it by hand. With the override, their explicit rule wins outright, then their mode, then the instance's verdict; and they may add anything by hand, because doing so is their explicit decision. """ if instance_rule is not None: instance_ok = instance_rule.effect == EFFECT_ALLOW instance_why: tuple[str, Rule | None] = (WHY_INSTANCE_RULE, instance_rule) elif not same_group: instance_ok, instance_why = False, (WHY_GROUP, None) else: instance_ok = instance_mode != MODE_CLOSED instance_why = (WHY_INSTANCE_CLOSED, None) if not override: if user_rule is not None: user_ok = user_rule.effect == EFFECT_ALLOW user_why: tuple[str, Rule | None] = (WHY_YOUR_RULE, user_rule) elif user_mode == MODE_CLOSED: user_ok, user_why = False, (WHY_YOUR_CLOSED, None) else: user_ok, user_why = True, ("", None) offered = instance_ok and user_ok why, rule = ("", None) if offered else (instance_why if not instance_ok else user_why) return Verdict(offered=offered, addable=instance_ok, why=why, rule=rule) if user_rule is not None: ok = user_rule.effect == EFFECT_ALLOW why, rule = (WHY_YOUR_RULE, user_rule) elif user_mode == MODE_CLOSED: ok, (why, rule) = False, (WHY_YOUR_CLOSED, None) elif user_mode == MODE_OPEN: allowed = instance_rule is not None and instance_rule.effect == EFFECT_ALLOW ok = same_group or allowed why, rule = (WHY_GROUP, None) else: ok = instance_ok why, rule = instance_why if ok: why, rule = "", None return Verdict(offered=ok, addable=True, why=why, rule=rule) # --- Loading what `decide` needs --------------------------------------------------- def _rules(db: DBSession, owner_id: str | None) -> list[Rule]: rows = db.scalars( select(TalkRule).where( TalkRule.owner_id.is_(None) if owner_id is None else TalkRule.owner_id == owner_id ) ) return [Rule(r.from_model, r.to_model, r.effect) for r in rows] def user_mode(user: User | None) -> str: if user is None: return "" value = str((user.settings_json or {}).get(SETTING_KEY) or "") return value if value in MODES else "" def may_override(db: DBSession, user: User | None) -> bool: from lembas.security import permissions return user is not None and permissions.has(db, user, PERMISSION) class Judge: """Everything `decide` needs for one person, loaded once per request. The model lists ask about every model they show; loading the rules and the connection groups per question would be a query per row of every picker. `user=None` is the instance's own view, for the admin page's matrix. """ def __init__(self, db: DBSession, user: User | None) -> None: from lembas.services import data_groups, settings_store self.instance_mode = settings_store.rules(db)["mode"] self.instance_rules = _rules(db, None) self.user_rules = _rules(db, user.id) if user is not None else [] self.user_mode = user_mode(user) self.override = may_override(db, user) self.groups = data_groups.connection_groups(db, user) self.default = data_groups.DEFAULT_GROUP def group_of(self, model: Model) -> str: return self.groups.get(model.connection_id, self.default) def verdict(self, main_model: str, main_group: str, target: Model) -> Verdict: return decide( instance_mode=self.instance_mode, instance_rule=match(self.instance_rules, main_model, target.model_id), user_mode=self.user_mode, user_rule=match(self.user_rules, main_model, target.model_id), override=self.override, same_group=self.group_of(target) == main_group, ) def candidates( db: DBSession, user: User | None, main_model: str, main_group: str ) -> list[tuple[Model, Verdict]]: """Every model this person can reach other than the main one, with its verdict.""" from lembas.services import chat as chat_service judge = Judge(db, user) return [ (model, judge.verdict(main_model, main_group, model)) for model in chat_service.available_models(db, user) if model.model_id != main_model ] def offered(db: DBSession, user: User | None, main_model: str, main_group: str) -> list[Model]: """The models a main model is offered on its own: the roster, a friend, the picker.""" found = candidates(db, user, main_model, main_group) return [model for model, verdict in found if verdict.offered] def addable(db: DBSession, user: User | None, main_model: str, main_group: str) -> list[Model]: """The models a person may add to a crowd by hand.""" found = candidates(db, user, main_model, main_group) return [model for model, verdict in found if verdict.addable] # --- Changing rules -------------------------------------------------------------------- def rules_of(db: DBSession, owner: User | None) -> list[TalkRule]: return list( db.scalars( select(TalkRule) .where( TalkRule.owner_id.is_(None) if owner is None else TalkRule.owner_id == owner.id ) .order_by(TalkRule.from_model, TalkRule.to_model) ) ) def set_rule( db: DBSession, owner: User | None, from_model: str, to_model: str, effect: str, *, both: bool = False, ) -> None: """Write one rule, or a pair in both directions. Replaces one for the same pair.""" from_model = (from_model or "").strip()[:300] or ANY_MODEL to_model = (to_model or "").strip()[:300] or ANY_MODEL if effect not in EFFECTS: effect = EFFECT_DENY pairs = [(from_model, to_model)] if both and from_model != to_model: pairs.append((to_model, from_model)) for frm, to in pairs: existing = db.scalar( select(TalkRule).where( (TalkRule.owner_id.is_(None) if owner is None else TalkRule.owner_id == owner.id), TalkRule.from_model == frm, TalkRule.to_model == to, ) ) if existing is not None: existing.effect = effect else: db.add( TalkRule( owner_id=owner.id if owner is not None else None, from_model=frm, to_model=to, effect=effect, ) ) db.commit() def delete_rule(db: DBSession, owner: User | None, rule_id: str) -> bool: """Remove one rule, only from the layer it belongs to.""" rule = db.get(TalkRule, rule_id) if rule is None or rule.owner_id != (owner.id if owner is not None else None): return False db.delete(rule) db.commit() return True def matrix(db: DBSession, user: User | None, models: list[Model]) -> list[dict]: """Main x target, drawn by the same `decide` that enforces the rules. Evaluated as if the main model's chat were in the main model's own group, which is what a chat started on it is. """ judge = Judge(db, user) rows = [] for main in models: group = judge.group_of(main) cells = [] for target in models: if target.model_id == main.model_id: cells.append(None) continue cells.append(judge.verdict(main.model_id, group, target)) rows.append({"main": main, "cells": cells}) return rows __all__ = [ "MODES", "MODE_CLOSED", "MODE_OPEN", "PERMISSION", "Judge", "Rule", "Verdict", "addable", "candidates", "decide", "delete_rule", "match", "matrix", "may_override", "offered", "rules_of", "set_rule", "user_mode", ]