[eric] mcp-gate: property-test the activation gate (forwarded => activated) + Z3 formal proof

backend/tests/formal/README.md

@@ -0,0 +1,20 @@
+# Formal proofs
+
+Machine-checked proofs of safety/security invariants that the unit/property
+tests can only *sample*. A property test tries thousands of cases; an SMT proof
+is exhaustive over the modeled domain (assert the negation, `unsat` => theorem).
+
+Not wired into prod or CI, and excluded from the packaged build (under `tests/`).
+Run manually:
+
+```
+pip install z3-solver
+python backend/tests/formal/mcp_gate_proof.py
+```
+
+- **`mcp_gate_proof.py`** , the MCP dispatch-gate invariant (`agent_manager._build_mcp_servers`):
+  a gated session forwards a server *only if* it was activated, an empty
+  activation list forwards zero, and a denied server is never forwarded. Sampled
+  by `tests/test_v2_invariants.py::test_mcp_gate_only_forwards_activated_servers`;
+  proven for all inputs here. The script also refutes a deliberately-buggy gate
+  (activation check dropped) so the proof can't be vacuous.

backend/tests/formal/mcp_gate_proof.py

@@ -0,0 +1,83 @@
+"""Formal proof (Z3 / SMT) of the MCP dispatch-gate security invariant.
+
+The product rule "MCP tools are reachable only after MCPActivate" is enforced at
+dispatch in agent_manager._build_mcp_servers: for a gated session a server is
+forwarded to the model only if its sanitized name is in session.active_mcps.
+
+tests/test_v2_invariants.py::test_mcp_gate_only_forwards_activated_servers
+SAMPLES that contract (400 random cases). This SMT proof is exhaustive over the
+modeled domain: we assert the negation of each property and ask Z3 for a
+counterexample. `unsat` means none can exist, so the property is a theorem,
+true for every possible input, not just the ones a test happened to try.
+
+Not wired into prod or CI. Run manually:
+    pip install z3-solver && python backend/tests/formal/mcp_gate_proof.py
+"""
+
+from z3 import And, Bool, Implies, Not, Or, Solver, sat, unsat
+
+
+def forwarded(installed, allowed, denied, active_is_none, active_t):
+    """Faithful model of the gate decision for one arbitrary server `t`
+    (agent_manager.py:165-203). A server ships to the model iff it is an
+    installed+configured MCP tool, passes the permission gate, isn't fully
+    denied, and EITHER the session is legacy (active_mcps is None) OR the
+    server is in active_mcps. Proving it for an arbitrary symbolic `t` proves
+    it for all servers."""
+    return And(installed, allowed, Not(denied), Or(active_is_none, active_t))
+
+
+def buggy_forwarded(installed, allowed, denied, active_is_none, active_t):
+    """The same gate with the activation check dropped, used to show the proof
+    has teeth: Z3 must be able to refute the no-leak property for this variant."""
+    return And(installed, allowed, Not(denied))
+
+
+def prove(name: str, claim) -> bool:
+    """`claim` should be valid (true for every input). Proven by showing its
+    negation is unsatisfiable."""
+    s = Solver()
+    s.add(Not(claim))
+    if s.check() == unsat:
+        print(f"  PROVED:  {name}")
+        return True
+    print(f"  FAILED:  {name}  counterexample: {s.model()}")
+    return False
+
+
+def main() -> None:
+    installed = Bool("installed")
+    allowed = Bool("allowed")
+    denied = Bool("denied")
+    active_is_none = Bool("active_is_none")  # legacy session (no activation gate)
+    active_t = Bool("active_t")              # server t is in active_mcps
+    fwd = forwarded(installed, allowed, denied, active_is_none, active_t)
+    gated = Not(active_is_none)
+
+    print("Proving MCP dispatch-gate invariants (exhaustive over all inputs):")
+    ok = True
+    # A. No leak: a gated session never forwards a non-activated server.
+    ok &= prove("gated  =>  (forwarded(t) -> activated(t))",
+                Implies(And(gated, fwd), active_t))
+    # B. Empty activation => zero servers (no t is active, so none ship).
+    ok &= prove("gated & !activated(t)  =>  !forwarded(t)",
+                Implies(And(gated, Not(active_t)), Not(fwd)))
+    # C. The permission gate still binds: a denied server is never forwarded.
+    ok &= prove("denied(t)  =>  !forwarded(t)", Implies(denied, Not(fwd)))
+
+    # Teeth: the buggy gate (activation check dropped) MUST be refutable, else
+    # the proof above would be vacuous.
+    print("Sanity-checking the proof has teeth (a buggy gate must be refuted):")
+    bug = buggy_forwarded(installed, allowed, denied, active_is_none, active_t)
+    s = Solver()
+    s.add(Not(Implies(And(gated, bug), active_t)))
+    assert s.check() == sat, "buggy gate should leak but Z3 couldn't refute it"
+    print(f"  REFUTED (as expected): a gate without the activation check leaks; "
+          f"counterexample = {s.model()}")
+
+    print("\nALL GATE PROPERTIES PROVED" if ok else "\nPROOF FAILED")
+    raise SystemExit(0 if ok else 1)
+
+
+if __name__ == "__main__":
+    main()

backend/tests/test_v2_invariants.py

@@ -550,6 +550,50 @@ def test_error_classify_gemini_resource_exhausted_is_transient():
     assert not _is_transient_capacity_error(Exception("403 permission denied"))
 
 
+@pytest.mark.asyncio
+async def test_mcp_gate_only_forwards_activated_servers():
+    """Dispatch-layer security invariant (the non-bypassable enforcement of
+    'MCP tools only via MCPActivate'): for a GATED session (active_mcps is a
+    list), _build_mcp_servers forwards ONLY servers whose sanitized name is in
+    active_mcps; an empty list forwards ZERO; None is the legacy all-allowed
+    path. The model cannot reach an unactivated server no matter what it asks
+    for. Property-checked over random installed sets and random activation
+    subsets, plus the two boundary cases."""
+    import random
+    from types import SimpleNamespace
+    from backend.apps.agents.agent_manager import AgentManager
+    mgr = AgentManager()
+    names = ["gmail", "drive", "slack", "reddit", "notion", "airtable"]
+
+    def installed():
+        return [SimpleNamespace(name=n, mcp_config={"x": 1}, enabled=True,
+                                auth_status="configured", auth_type="apikey") for n in names]
+
+    # allowed_tools == get_all_tool_names() bypasses the (separate) permission
+    # gate so we isolate the ACTIVATION gate. _sanitize_server_name -> identity.
+    with patch("backend.apps.agents.agent_manager.load_all_tools", side_effect=installed), \
+         patch("backend.apps.agents.agent_manager.get_all_tool_names", return_value=["__ALL__"]), \
+         patch("backend.apps.agents.agent_manager._sanitize_server_name", side_effect=lambda n: n), \
+         patch("backend.apps.agents.agent_manager._is_fully_denied", return_value=False), \
+         patch("backend.apps.agents.agent_manager.derive_mcp_config", side_effect=lambda t: {"command": "x"}):
+        allowed = ["__ALL__"]
+        # Boundary 1: empty activation list -> zero servers, always.
+        assert await mgr._build_mcp_servers(allowed, active_mcps=[]) == {}
+        # Boundary 2: None (legacy) -> permission gate only, all forwarded.
+        assert set((await mgr._build_mcp_servers(allowed, active_mcps=None)).keys()) == set(names)
+        # Property: forwarded set is ALWAYS a subset of the activated set, and
+        # equals exactly the activated-and-installed intersection.
+        rng = random.Random(1234)
+        for _ in range(400):
+            active = rng.sample(names, rng.randint(0, len(names)))
+            # throw in a bogus name the gate must never invent a server for
+            if rng.random() < 0.3:
+                active = active + ["ghost-not-installed"]
+            forwarded = set((await mgr._build_mcp_servers(allowed, active_mcps=active)).keys())
+            assert forwarded <= set(active), f"leaked {forwarded - set(active)} for active={active}"
+            assert forwarded == (set(active) & set(names)), f"mismatch for active={active}"
+
+
 def test_banned_models_not_offered():
     """Claude Fable (banned) and Gemini 3.1 Pro (no working lane: AG can't serve
     it, AI Studio key 429s pro-preview) were pulled from the picker. Guard so a