V2 P0: Memory Reliability Gate + MCP cortex_evaluate (#71)

api/main.py

@@ -29,6 +29,7 @@
 from api.evidence import router as evidence_router
 from api.gdpr import router as gdpr_router
 from api.memory_state import router as memory_state_router
+from api.reliability import router as reliability_router
 from api.metrics import record_http_request, record_query, render_metrics
 from api.decisions import router as decisions_router
 from api.deps import RolesDep, memory, set_memory_service
@@ -109,6 +110,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 app.include_router(remember_router)
 app.include_router(evidence_router)
 app.include_router(memory_state_router)
+app.include_router(reliability_router)
 
 
 @app.middleware("http")

api/reliability.py

@@ -0,0 +1,62 @@
+"""POST /reliability/evaluate — Memory Reliability Gate API."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+
+from api.deps import RolesDep
+from reliability.gate import ReliabilityGate
+from shared.v2_models import ActionRisk, ReliabilityVerdict
+
+router = APIRouter(prefix="/reliability", tags=["reliability"])
+
+
+class EvaluateRequest(BaseModel):
+    workspace_id: str
+    query: str = Field(min_length=3, max_length=2000)
+    candidate_action: dict[str, Any] = Field(default_factory=dict)
+    memory_ids: list[str] = Field(default_factory=list)
+    claims: list[dict[str, Any]] = Field(default_factory=list)
+    memories: list[dict[str, Any]] = Field(default_factory=list)
+    risk: ActionRisk | None = None
+
+
+class EvaluateResponse(BaseModel):
+    decision: ReliabilityVerdict
+    memory_confidence: float
+    action_confidence: float
+    risk: ActionRisk
+    reason_codes: list[str]
+    required_checks: list[str]
+    id: str
+
+
+@router.post(
+    "/evaluate",
+    response_model=EvaluateResponse,
+    summary="Evaluate whether memory is reliable enough for an agent action",
+)
+def evaluate_reliability(body: EvaluateRequest, roles: RolesDep) -> EvaluateResponse:
+    """Run the Memory Reliability Gate."""
+    gate = ReliabilityGate()
+    result = gate.evaluate(
+        workspace_id=body.workspace_id,
+        query=body.query,
+        candidate_action=body.candidate_action,
+        claims=body.claims,
+        memories=body.memories,
+        caller_roles=roles,
+        risk=body.risk,
+    )
+    return EvaluateResponse(
+        decision=result.decision,
+        memory_confidence=result.memory_confidence,
+        action_confidence=result.action_confidence,
+        risk=result.risk,
+        reason_codes=result.reason_codes,
+        required_checks=result.required_checks,
+        id=result.id,
+    )

mcp/server.js

@@ -125,6 +125,46 @@ server.tool(
   },
 );
 
+server.tool(
+  "cortex_evaluate",
+  {
+    description:
+      "Evaluate whether retrieved memory is reliable enough for a candidate agent action (Reliability Gate)",
+    inputSchema: {
+      type: "object",
+      properties: {
+        workspace_id: { type: "string" },
+        query: { type: "string" },
+        candidate_action: { type: "object" },
+        claims: { type: "array", items: { type: "object" } },
+        memories: { type: "array", items: { type: "object" } },
+        risk: { type: "string" },
+      },
+      required: ["workspace_id", "query"],
+    },
+  },
+  async ({
+    workspace_id,
+    query,
+    candidate_action = {},
+    claims = [],
+    memories = [],
+    risk,
+  }) => {
+    const payload = await postJson("/reliability/evaluate", {
+      workspace_id,
+      query,
+      candidate_action,
+      claims,
+      memories,
+      risk,
+    });
+    return {
+      content: [{ type: "text", text: JSON.stringify(payload, null, 2) }],
+    };
+  },
+);
+
 async function main() {
   const transport = new StdioServerTransport();
   await server.connect(transport);

reliability/__init__.py

@@ -0,0 +1,5 @@
+"""Reliability package — Memory Reliability Gate."""
+
+from reliability.gate import ReliabilityGate
+
+__all__ = ["ReliabilityGate"]

reliability/confidence.py

@@ -0,0 +1,27 @@
+"""Confidence aggregation for reliability gate."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def memory_confidence_from_claims(claims: list[dict[str, Any]]) -> float:
+    """Aggregate memory confidence from claim payloads."""
+    if not claims:
+        return 0.0
+    scores = [float(c.get("confidence") or 0.0) for c in claims]
+    return sum(scores) / len(scores)
+
+
+def action_confidence(
+    *,
+    memory_conf: float,
+    risk_penalty: float,
+    conflict_penalty: float,
+    auth_ok: bool,
+) -> float:
+    """Derive action confidence from memory conf and penalties."""
+    if not auth_ok:
+        return min(memory_conf, 0.15)
+    value = memory_conf - risk_penalty - conflict_penalty
+    return max(0.0, min(1.0, value))

reliability/gate.py

@@ -0,0 +1,167 @@
+"""Memory Reliability Gate — decide ACT/VERIFY/ASK/ESCALATE/BLOCK."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import structlog
+
+from reliability import reason_codes as rc
+from reliability.confidence import action_confidence, memory_confidence_from_claims
+from reliability.policies import roles_may_act, threshold_for, VERIFY_BAND
+from shared.v2_models import ActionRisk, ReliabilityDecision, ReliabilityVerdict
+
+log = structlog.get_logger(__name__)
+
+_RISK_PENALTY: dict[str, float] = {
+    "READ_ONLY": 0.0,
+    "LOW_RISK_WRITE": 0.05,
+    "REVERSIBLE_WRITE": 0.10,
+    "HIGH_IMPACT_WRITE": 0.20,
+    "IRREVERSIBLE": 0.30,
+    "PRIVILEGED": 0.35,
+}
+
+
+class ReliabilityGate:
+    """Policy-aware gate between retrieval and agent action."""
+
+    def evaluate(
+        self,
+        *,
+        workspace_id: str,
+        query: str,
+        candidate_action: dict[str, Any],
+        claims: list[dict[str, Any]] | None = None,
+        memories: list[dict[str, Any]] | None = None,
+        caller_roles: list[str] | None = None,
+        risk: ActionRisk | str | None = None,
+    ) -> ReliabilityDecision:
+        """Evaluate whether memory is reliable enough for the candidate action."""
+        claims = claims or []
+        memories = memories or []
+        roles = caller_roles or ["authenticated"]
+        action_risk: ActionRisk = (risk or candidate_action.get("risk") or "READ_ONLY")  # type: ignore[assignment]
+        if action_risk not in _RISK_PENALTY:
+            action_risk = "HIGH_IMPACT_WRITE"
+
+        reasons: list[str] = []
+        checks: list[str] = []
+
+        mem_conf = memory_confidence_from_claims(claims) if claims else 0.0
+        if memories and not claims:
+            # Fallback: use decision trust/importance averages
+            scores = [
+                float(m.get("trust_score") or m.get("importance_score") or 0.0)
+                for m in memories
+            ]
+            mem_conf = sum(scores) / len(scores) if scores else 0.0
+
+        if mem_conf < 0.4:
+            reasons.append(rc.LOW_MEMORY_CONFIDENCE)
+        if not claims and not memories:
+            reasons.append(rc.INSUFFICIENT_EVIDENCE)
+
+        conflict = False
+        for c in claims:
+            if str(c.get("status")) == "QUARANTINED":
+                reasons.append(rc.QUARANTINED_MEMORY)
+            if str(c.get("status")) == "SUPERSEDED":
+                reasons.append(rc.SUPERSEDED_MEMORY)
+            if str(c.get("assertion_type")) == "INFERRED":
+                reasons.append(rc.INFERRED_ONLY)
+            if c.get("conflict"):
+                conflict = True
+                reasons.append(rc.HIGH_AUTHORITY_CONFLICT)
+
+        # Heuristic: single low-authority claim
+        if len(claims) == 1 and float(claims[0].get("confidence") or 0) < 0.7:
+            reasons.append(rc.SINGLE_SOURCE_SUPPORT)
+
+        # Stale procedure signal from action metadata
+        if candidate_action.get("stale_procedure") or candidate_action.get("procedure_status") == "STALE":
+            reasons.append(rc.STALE_PROCEDURE)
+            checks.append("verify current procedure version")
+        if candidate_action.get("active_migration"):
+            reasons.append(rc.ACTIVE_MIGRATION)
+            checks.append("confirm migration window with owner")
+
+        auth_ok = roles_may_act(roles, action_risk)
+        if not auth_ok:
+            reasons.append(rc.ACTION_NOT_AUTHORIZED)
+
+        conflict_penalty = 0.25 if conflict or rc.HIGH_AUTHORITY_CONFLICT in reasons else 0.0
+        if rc.STALE_PROCEDURE in reasons:
+            conflict_penalty += 0.15
+        if rc.ACTIVE_MIGRATION in reasons:
+            conflict_penalty += 0.20
+
+        act_conf = action_confidence(
+            memory_conf=mem_conf,
+            risk_penalty=_RISK_PENALTY.get(action_risk, 0.2),
+            conflict_penalty=conflict_penalty,
+            auth_ok=auth_ok,
+        )
+
+        verdict = self._decide(
+            action_risk=action_risk,
+            mem_conf=mem_conf,
+            act_conf=act_conf,
+            reasons=reasons,
+            auth_ok=auth_ok,
+        )
+
+        # Dedupe reasons preserving order
+        seen: set[str] = set()
+        uniq_reasons = []
+        for r in reasons:
+            if r not in seen:
+                seen.add(r)
+                uniq_reasons.append(r)
+
+        decision = ReliabilityDecision(
+            workspace_id=workspace_id,
+            query_id=None,
+            candidate_action={"query": query, **candidate_action},
+            memory_confidence=round(mem_conf, 4),
+            action_confidence=round(act_conf, 4),
+            risk=action_risk,
+            decision=verdict,
+            reason_codes=uniq_reasons,
+            required_checks=checks,
+        )
+        log.info(
+            "reliability_gate.evaluated",
+            decision=verdict,
+            risk=action_risk,
+            memory_confidence=decision.memory_confidence,
+            action_confidence=decision.action_confidence,
+            reasons=uniq_reasons,
+        )
+        return decision
+
+    @staticmethod
+    def _decide(
+        *,
+        action_risk: ActionRisk,
+        mem_conf: float,
+        act_conf: float,
+        reasons: list[str],
+        auth_ok: bool,
+    ) -> ReliabilityVerdict:
+        if not auth_ok or rc.ACTION_NOT_AUTHORIZED in reasons:
+            return "BLOCK"
+        if rc.QUARANTINED_MEMORY in reasons or rc.ACTIVE_MIGRATION in reasons:
+            return "BLOCK"
+        if rc.INSUFFICIENT_EVIDENCE in reasons or mem_conf < 0.35:
+            return "ASK"
+        if rc.HIGH_AUTHORITY_CONFLICT in reasons:
+            return "ESCALATE"
+        thr = threshold_for(action_risk)
+        if act_conf >= thr:
+            return "ACT"
+        if act_conf >= thr - VERIFY_BAND or rc.STALE_PROCEDURE in reasons:
+            return "VERIFY"
+        if act_conf < 0.35:
+            return "ASK"
+        return "ESCALATE"

reliability/policies.py

@@ -0,0 +1,32 @@
+"""Action risk levels and policy thresholds."""
+
+from __future__ import annotations
+
+from shared.v2_models import ActionRisk
+
+# Minimum action_confidence required to ACT, by risk.
+ACT_THRESHOLDS: dict[ActionRisk, float] = {
+    "READ_ONLY": 0.40,
+    "LOW_RISK_WRITE": 0.55,
+    "REVERSIBLE_WRITE": 0.70,
+    "HIGH_IMPACT_WRITE": 0.85,
+    "IRREVERSIBLE": 0.92,
+    "PRIVILEGED": 0.95,
+}
+
+VERIFY_BAND = 0.15  # if within this of threshold → VERIFY instead of ACT
+
+
+def threshold_for(risk: ActionRisk | str) -> float:
+    """Return ACT threshold for a risk level."""
+    return float(ACT_THRESHOLDS.get(risk, 0.85))  # type: ignore[arg-type]
+
+
+def roles_may_act(roles: list[str], risk: ActionRisk | str) -> bool:
+    """Authorization check outside the LLM."""
+    role_set = {r.lower() for r in roles}
+    if risk in {"HIGH_IMPACT_WRITE", "IRREVERSIBLE", "PRIVILEGED"}:
+        return bool(role_set & {"admin", "operator", "sre", "lead"})
+    if risk in {"REVERSIBLE_WRITE", "LOW_RISK_WRITE"}:
+        return bool(role_set & {"admin", "operator", "sre", "lead", "engineer", "authenticated"})
+    return "authenticated" in role_set or bool(role_set)

reliability/reason_codes.py

@@ -0,0 +1,14 @@
+"""Reliability reason codes."""
+
+from __future__ import annotations
+
+STALE_PROCEDURE = "STALE_PROCEDURE"
+SINGLE_SOURCE_SUPPORT = "SINGLE_SOURCE_SUPPORT"
+HIGH_AUTHORITY_CONFLICT = "HIGH_AUTHORITY_CONFLICT"
+ACTION_NOT_AUTHORIZED = "ACTION_NOT_AUTHORIZED"
+INSUFFICIENT_EVIDENCE = "INSUFFICIENT_EVIDENCE"
+QUARANTINED_MEMORY = "QUARANTINED_MEMORY"
+SUPERSEDED_MEMORY = "SUPERSEDED_MEMORY"
+LOW_MEMORY_CONFIDENCE = "LOW_MEMORY_CONFIDENCE"
+ACTIVE_MIGRATION = "ACTIVE_MIGRATION"
+INFERRED_ONLY = "INFERRED_ONLY"