Spaces:
Running
Running
| """ | |
| core/relationship_graph.py — Role-Scoped Context Delivery | |
| Instead of broadcasting everything to everyone, the relationship graph | |
| pre-filters what each role receives. Prevents token explosion, accidental | |
| consensus loops, and scope creep. | |
| Default roles ship pre-configured. Add/update at runtime via the API. | |
| """ | |
| from __future__ import annotations | |
| from typing import Dict, List, Optional, Set | |
| # ── Default role → frequency patterns ──────────────────────────────────────── | |
| # Values are prefix/substring patterns matched against frequency names. | |
| # Historian receives everything — it's the only role that does. | |
| _DEFAULT_ROLES: Dict[str, List[str]] = { | |
| "CodeAgent": ["architecture", "api", "backend", "test_fail", "commit", | |
| "router", "service", "database", "schema"], | |
| "FrontendAgent": ["ui", "frontend", "component", "design", "css", | |
| "ux", "layout", "visual"], | |
| "QAAgent": ["test", "commit", "fail", "coverage", "regression", | |
| "bug", "error", "assert"], | |
| "ResearchAgent": ["requirement", "spec", "open_question", "rfc", | |
| "external", "arxiv", "paper", "research"], | |
| "SecurityAgent": ["security", "auth", "secret", "key", "credential", | |
| "vuln", "cve", "owasp"], | |
| "InfraAgent": ["infra", "deploy", "docker", "k8s", "ci", "cd", | |
| "pipeline", "release"], | |
| "DataAgent": ["data", "schema", "migration", "etl", "pipeline", | |
| "warehouse", "feed", "rss"], | |
| "Historian": ["*"], # receives everything | |
| } | |
| class RelationshipGraph: | |
| """ | |
| Role → frequency filter. tune_in(role=) uses this to scope context delivery. | |
| Dumb and fast — no LLM, no scoring, just pattern matching. | |
| """ | |
| def __init__(self): | |
| self._roles: Dict[str, List[str]] = dict(_DEFAULT_ROLES) | |
| # ── Filtering ───────────────────────────────────────────────────────────── | |
| def allowed_frequencies(self, role: str, available: List[str]) -> List[str]: | |
| """Return which frequencies from `available` this role may receive.""" | |
| patterns = self._roles.get(role, []) | |
| if not patterns: | |
| return [] | |
| if "*" in patterns: | |
| return available | |
| result: List[str] = [] | |
| for freq in available: | |
| fl = freq.lower() | |
| if any(p.lower() in fl for p in patterns): | |
| result.append(freq) | |
| return result | |
| def filter_signals(self, role: str, signals: List[dict]) -> List[dict]: | |
| """Filter a list of already-tuned signals to those allowed for role.""" | |
| if not role: | |
| return signals | |
| patterns = self._roles.get(role, []) | |
| if not patterns: | |
| return [] | |
| if "*" in patterns: | |
| return signals | |
| return [ | |
| s for s in signals | |
| if any(p.lower() in s.get("name", "").lower() for p in patterns) | |
| ] | |
| # ── Management ──────────────────────────────────────────────────────────── | |
| def set_role(self, role: str, patterns: List[str]): | |
| self._roles[role] = patterns | |
| def get_role(self, role: str) -> Optional[List[str]]: | |
| return self._roles.get(role) | |
| def all_roles(self) -> Dict[str, List[str]]: | |
| return dict(self._roles) | |
| def remove_role(self, role: str) -> bool: | |
| if role in self._roles: | |
| del self._roles[role] | |
| return True | |
| return False | |
| relationship_graph = RelationshipGraph() | |