brain: core v0 (guardian-gated model calls + no-claim post + provenance)

backend/core/action_router.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+from backend.core.crit_contract import Decision
+from backend.core.crit_gate import crit_gate
+from backend.core.model_gateway import ModelGateway
+
+
+class ActionRouter:
+    def __init__(self, model_gateway: ModelGateway) -> None:
+        self.model_gateway = model_gateway
+
+    def route(
+        self,
+        action: str,
+        context: Dict[str, Any],
+        flags: Dict[str, Any],
+        *,
+        decision: Decision | None = None,
+    ) -> Optional[str]:
+        if decision is None:
+            decision = crit_gate(
+                {"action": action, "intent": flags.get("intent", ""), "context": context, "flags": flags}
+            )
+        if decision.status in {"BLOCK", "REQUIRE_CONFIRMATION"} or decision.requires_human_confirmation:
+            return None
+        model_name = flags.get("model", "stub")
+        return self.model_gateway.call(action, model=model_name)

backend/core/brain_core.py

@@ -1,58 +1,98 @@
 from __future__ import annotations
 
+from backend.core.action_router import ActionRouter
 from backend.core.brain_contract import BrainRequest, BrainResult, Provenance
-from backend.core.guardian_engine import GuardianEngine
+from backend.core.crit_gate import crit_gate
+from backend.core.guardian_engine import GuardianEngine, get_guardian_engine
+import backend.core.memory_gate as memory_gate
 from backend.core.model_gateway import ModelGateway
 from backend.core.response_postprocess import apply as apply_postprocess
 
 
-class BrainCore:
-    def __init__(self, guardian_engine: GuardianEngine, model_gateway: ModelGateway) -> None:
-        self.guardian_engine = guardian_engine
-        self.model_gateway = model_gateway
+def run_brain(
+    action: str,
+    intent: str = "",
+    context: dict | None = None,
+    flags: dict | None = None,
+    *,
+    guardian_engine: GuardianEngine | None = None,
+    model_gateway: ModelGateway | None = None,
+    action_router: ActionRouter | None = None,
+) -> BrainResult:
+    ctx = context or {}
+    flg = flags or {}
+    engine = guardian_engine or get_guardian_engine()
+    gateway = model_gateway or ModelGateway()
+    router = action_router or ActionRouter(gateway)
 
-    def run(self, request: BrainRequest) -> BrainResult:
-        decision = self.guardian_engine.evaluate(
-            request.text,
-            intent=request.intent,
-            context=request.context,
-            flags=request.flags,
-        )
-        lane = request.flags.get("lane", "default")
-        hit_rule_ids = [h.rule_id for h in decision.hits]
-        provenance = Provenance(
-            lane=lane,
-            decision_status=decision.status,
-            hit_rule_ids=hit_rule_ids,
-            model_used=None,
-            post_rewrites=[],
+    # CritGate es el front-door gate; delega en GuardianEngine.
+    decision = crit_gate(
+        {"action": action, "intent": intent, "context": ctx, "flags": flg},
+        engine=engine,
+    )
+
+    lane = flg.get("lane", "default")
+    hit_rule_ids = [h.rule_id for h in decision.hits]
+    provenance = Provenance(
+        lane=lane,
+        decision_status=decision.status,
+        hit_rule_ids=hit_rule_ids,
+        model_used=None,
+        post_rewrites=[],
+    )
+
+    if decision.status in {"BLOCK", "REQUIRE_CONFIRMATION"} or decision.requires_human_confirmation:
+        return BrainResult(
+            status=decision.status,
+            assistant_message=decision.assistant_message,
+            requires_human_confirmation=decision.requires_human_confirmation,
+            provenance=provenance,
         )
 
-        if decision.status == "BLOCK":
+    if ctx.get("memory_consent") and ctx.get("memory_write") and ctx.get("memory_content"):
+        gate_decision = memory_gate.evaluate_memory_write(
+            ctx.get("user_id", "unknown"),
+            ctx.get("memory_content", ""),
+            context=ctx,
+        ).decision
+        if gate_decision.status == "BLOCK":
             return BrainResult(
-                status=decision.status,
-                assistant_message=decision.assistant_message,
-                requires_human_confirmation=decision.requires_human_confirmation,
+                status=gate_decision.status,
+                assistant_message=gate_decision.assistant_message,
+                requires_human_confirmation=gate_decision.requires_human_confirmation,
                 provenance=provenance,
             )
-
-        if decision.requires_human_confirmation:
-            return BrainResult(
-                status=decision.status,
-                assistant_message=decision.assistant_message,
-                requires_human_confirmation=True,
-                provenance=provenance,
+        if ctx.get("memory_semantic_opt_in"):
+            _ = memory_gate.evaluate_memory_index(
+                ctx.get("user_id", "unknown"),
+                ctx.get("memory_content", ""),
+                context=ctx,
             )
 
-        model_name = request.flags.get("model", "stub")
-        raw = self.model_gateway.call(request.text, model=model_name)
-        rewritten, rewrites = apply_postprocess(raw, decision, request.context)
-        provenance.post_rewrites.extend(rewrites)
-        provenance.model_used = model_name
+    raw = router.route(action, ctx, flg, decision=decision) or ""
+    final, rewrites = apply_postprocess(raw, decision, ctx)
+    provenance.post_rewrites.extend(rewrites)
+    provenance.model_used = flg.get("model", "stub")
 
-        return BrainResult(
-            status=decision.status,
-            assistant_message=rewritten,
-            requires_human_confirmation=decision.requires_human_confirmation,
-            provenance=provenance,
+    return BrainResult(
+        status=decision.status,
+        assistant_message=final,
+        requires_human_confirmation=decision.requires_human_confirmation,
+        provenance=provenance,
+    )
+
+
+class BrainCore:
+    def __init__(self, guardian_engine: GuardianEngine, model_gateway: ModelGateway) -> None:
+        self.guardian_engine = guardian_engine
+        self.model_gateway = model_gateway
+
+    def run(self, request: BrainRequest) -> BrainResult:
+        return run_brain(
+            action=request.text,
+            intent=request.intent,
+            context=request.context,
+            flags=request.flags,
+            guardian_engine=self.guardian_engine,
+            model_gateway=self.model_gateway,
         )

backend/core/crit_contract.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Literal
+
+# Canonical status values returned by the engine
+Status = Literal["ALLOW", "WARN", "REQUIRE_CONFIRMATION", "BLOCK"]
+
+# Contract aliases used in tests/pipeline mapping
+ALLOW = "ALLOW"
+WARN = "WARN"
+REQUIRE_CONFIRMATION = "REQUIRE_CONFIRMATION"
+BLOCK = "BLOCK"
+
+DENY = "DENY"
+CLARIFY = "CLARIFY"
+NEEDS_HUMAN = "NEEDS_HUMAN"
+REWRITE_OR_CLARIFY = "REWRITE_OR_CLARIFY"
+
+
+@dataclass
+class RuleHit:
+    rule_id: str
+    category: str
+    severity: str
+    message: str
+    enforcement: str
+
+
+@dataclass
+class Decision:
+    status: Status
+    hits: List[RuleHit] = field(default_factory=list)
+    requires_human_confirmation: bool = False
+    assistant_message: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)

backend/core/crit_gate.py

@@ -10,7 +10,8 @@
 
 from typing import Any, Dict
 
-from backend.core.guardian_engine import Decision, GuardianEngine, get_guardian_engine
+from backend.core.crit_contract import Decision
+from backend.core.guardian_engine import GuardianEngine, get_guardian_engine
 
 
 # Estado legacy para importadores antiguos; preferir Decision.status

backend/core/guardian_engine.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 from typing import Any, Dict, List, Literal, Optional
 
+from backend.core.crit_contract import Decision, RuleHit, Status
 from backend.core.guardian_telemetry import get_ruleset_version, get_telemetry_sink
 from backend.core.response_postprocess import apply as apply_postprocess
 
@@ -18,26 +19,6 @@
 
 logger = logging.getLogger(__name__)
 
-# Canonical status values returned by the engine
-Status = Literal["ALLOW", "WARN", "REQUIRE_CONFIRMATION", "BLOCK"]
-
-
-@dataclass
-class RuleHit:
-    rule_id: str
-    category: str
-    severity: str
-    message: str
-    enforcement: str
-
-
-@dataclass
-class Decision:
-    status: Status
-    hits: List[RuleHit] = field(default_factory=list)
-    requires_human_confirmation: bool = False
-    assistant_message: str = ""
-    metadata: Dict[str, Any] = field(default_factory=dict)
 
 
 class GuardianRulesError(RuntimeError):

backend/core/memory_gate.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from typing import Any, Dict, Optional
 
-from backend.core.guardian_engine import Decision, RuleHit
+from backend.core.crit_contract import Decision, RuleHit
 from backend.core.memory_store import MemoryStore
 
 

docs/BRAIN_CORE_SINGLE_SOURCE.md

@@ -0,0 +1,18 @@
+# brain core single source of truth
+
+## pipeline (ssot)
+CRITGate -> GuardianEngine (rules) -> MemoryGate (opt-in) -> ActionRouter -> Telemetry (opt-in)
+
+## authoritative path
+- unico pipeline valido para llamadas a modelo
+- no hay puertas paralelas ni duplicadas
+
+## critgate placement
+- vive en `backend/core/crit_gate.py`
+- es el front-door gate: rapido, barato, sin reglas detalladas
+
+## non-goals
+- no duplicar guardianes
+- no shadow gates
+- no postprocess duplicado
+

docs/CRITGATE_SINGLE_SOURCE.md

@@ -0,0 +1,16 @@
+# critgate single source of truth
+
+## que es
+critgate es la unica puerta obligatoria para toda accion (brain, routers, tools, memory, model calls).
+
+## contrato unico
+`crit_gate({action, intent, context, flags}) -> Decision`
+
+## pipeline
+CritGate -> GuardianEngine (rules) -> MemoryGate (opt-in) -> ActionRouter -> Telemetry (opt-in)
+
+## invariantes
+- no hay engines duplicados
+- no hay gates paralelos
+- no prompt raw en telemetria por defecto
+