From 6262aad0f38285ab6e559cba3296da216aaf7bdb Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 19 Sep 2026 00:02:04 +0000 Subject: [PATCH] V2 P0: Memory Reliability Gate API and MCP cortex_evaluate (#71). Add reliability package (gate, policies, confidence, reason codes), POST /reliability/evaluate, signature BLOCK scenario tests, and MCP tool. Co-authored-by: Abhinaysai Kamineni --- api/main.py | 2 + api/reliability.py | 62 ++++++++++++ mcp/server.js | 40 ++++++++ reliability/__init__.py | 5 + reliability/confidence.py | 27 ++++++ reliability/gate.py | 167 +++++++++++++++++++++++++++++++++ reliability/policies.py | 32 +++++++ reliability/reason_codes.py | 14 +++ tests/reliability/test_gate.py | 89 ++++++++++++++++++ 9 files changed, 438 insertions(+) create mode 100644 api/reliability.py create mode 100644 reliability/__init__.py create mode 100644 reliability/confidence.py create mode 100644 reliability/gate.py create mode 100644 reliability/policies.py create mode 100644 reliability/reason_codes.py create mode 100644 tests/reliability/test_gate.py diff --git a/api/main.py b/api/main.py index 0a78d3f..ad5afea 100644 --- a/api/main.py +++ b/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") diff --git a/api/reliability.py b/api/reliability.py new file mode 100644 index 0000000..fca57d0 --- /dev/null +++ b/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, + ) diff --git a/mcp/server.js b/mcp/server.js index 196acf1..800548a 100644 --- a/mcp/server.js +++ b/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); diff --git a/reliability/__init__.py b/reliability/__init__.py new file mode 100644 index 0000000..155ed20 --- /dev/null +++ b/reliability/__init__.py @@ -0,0 +1,5 @@ +"""Reliability package — Memory Reliability Gate.""" + +from reliability.gate import ReliabilityGate + +__all__ = ["ReliabilityGate"] diff --git a/reliability/confidence.py b/reliability/confidence.py new file mode 100644 index 0000000..18fe3bd --- /dev/null +++ b/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)) diff --git a/reliability/gate.py b/reliability/gate.py new file mode 100644 index 0000000..3c711a5 --- /dev/null +++ b/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" diff --git a/reliability/policies.py b/reliability/policies.py new file mode 100644 index 0000000..be9decd --- /dev/null +++ b/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) diff --git a/reliability/reason_codes.py b/reliability/reason_codes.py new file mode 100644 index 0000000..482b3a1 --- /dev/null +++ b/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" diff --git a/tests/reliability/test_gate.py b/tests/reliability/test_gate.py new file mode 100644 index 0000000..1a425b7 --- /dev/null +++ b/tests/reliability/test_gate.py @@ -0,0 +1,89 @@ +"""Tests for Memory Reliability Gate.""" + +from __future__ import annotations + +from reliability.gate import ReliabilityGate + + +def test_same_memory_different_risk() -> None: + gate = ReliabilityGate() + claims = [{"confidence": 0.8, "status": "ACTIVE", "assertion_type": "ASSERTED"}] + low = gate.evaluate( + workspace_id="ws", + query="status?", + candidate_action={"type": "read"}, + claims=claims, + caller_roles=["authenticated"], + risk="READ_ONLY", + ) + high = gate.evaluate( + workspace_id="ws", + query="restart?", + candidate_action={"type": "restart_service", "target": "payments"}, + claims=claims, + caller_roles=["authenticated"], + risk="IRREVERSIBLE", + ) + assert low.decision in {"ACT", "VERIFY"} + assert high.decision != "ACT" + assert high.action_confidence < low.action_confidence + + +def test_unauthorized_blocks_privileged() -> None: + gate = ReliabilityGate() + result = gate.evaluate( + workspace_id="ws", + query="restart payments", + candidate_action={"type": "restart_service"}, + claims=[{"confidence": 0.99, "status": "ACTIVE"}], + caller_roles=["authenticated"], + risk="PRIVILEGED", + ) + assert result.decision == "BLOCK" + assert "ACTION_NOT_AUTHORIZED" in result.reason_codes + + +def test_signature_payments_restart_block() -> None: + """Signature V2 scenario: high memory conf but BLOCK action.""" + gate = ReliabilityGate() + claims = [ + { + "confidence": 0.95, + "status": "ACTIVE", + "assertion_type": "ASSERTED", + "conflict": True, + }, + {"confidence": 0.9, "status": "SUPERSEDED", "assertion_type": "ASSERTED"}, + ] + result = gate.evaluate( + workspace_id="local-dev", + query="Should I restart the payments service?", + candidate_action={ + "type": "restart_service", + "target": "payments-service", + "stale_procedure": True, + "active_migration": True, + }, + claims=claims, + caller_roles=["authenticated"], + risk="HIGH_IMPACT_WRITE", + ) + assert result.decision == "BLOCK" + assert result.memory_confidence >= 0.9 + assert result.action_confidence < 0.5 + assert "ACTION_NOT_AUTHORIZED" in result.reason_codes or "ACTIVE_MIGRATION" in result.reason_codes + + +def test_insufficient_evidence_asks() -> None: + gate = ReliabilityGate() + result = gate.evaluate( + workspace_id="ws", + query="unknown domain", + candidate_action={}, + claims=[], + memories=[], + caller_roles=["admin"], + risk="READ_ONLY", + ) + assert result.decision == "ASK" + assert "INSUFFICIENT_EVIDENCE" in result.reason_codes