[eric] phase2: suggest-only offer_for_gated_server core + gate-safety invariants (in-task MCP connect)

This commit is contained in:
ciregenz
2026-06-16 22:23:02 -07:00
parent 177d2b6fb7
commit b8dc125da7
2 changed files with 78 additions and 0 deletions
+14
View File
@@ -138,6 +138,20 @@ def _build_available_shortlist(settings) -> list[CuratedEntry]:
]
def offer_for_gated_server(server_name: str, settings) -> CuratedEntry | None:
"""Mid-run a running agent may reach for a vetted MCP it isn't granted; this maps that
server to a one-click connect offer to SHOW the user. Suggest-only by construction: it
returns data to display, never an action that grants access, so it cannot widen the MCP
surface (activation stays behind MCPActivate + the dispatch gate). Returns None unless the
server is vetted AND inactive AND not dismissed, reusing the same filter as the preflight."""
if not server_name or not isinstance(server_name, str):
return None
entry = next((e for e in _build_available_shortlist(settings) if e["id"] == server_name), None)
if entry is None:
return None
return {"id": entry["id"], "title": entry["title"], "description": entry["description"], "reason": ""}
def _decorate(llm_suggestion: dict, available: list[CuratedEntry]) -> dict | None:
"""Expand an LLM-returned {id, reason} into the full frontend shape."""
entry = next((e for e in available if e["id"] == llm_suggestion["id"]), None)
+64
View File
@@ -0,0 +1,64 @@
"""Gate-safety invariants for the Phase 2 in-task connect offer (offer_for_gated_server).
The whole point of the offer is that it can ONLY ever suggest, never grant: it must surface a
vetted, inactive, not-dismissed MCP for the user to one-click-connect, and it must never carry
anything that could widen the MCP surface on its own. These tests make a bad offer state fail
loudly instead of shipping a silent gate bypass.
"""
from types import SimpleNamespace
import backend.apps.agents.core.mcp_preflight as pf
from backend.apps.agents.core.mcp_preflight import (
CURATED_SHORTLIST,
offer_for_gated_server,
)
VETTED = {e["id"] for e in CURATED_SHORTLIST}
OFFER_SHAPE = {"id", "title", "description", "reason"}
def _settings(dismissed=None):
return SimpleNamespace(dismissed_mcp_suggestions=dismissed or {})
def test_offer_only_returns_vetted_inactive(monkeypatch):
monkeypatch.setattr(pf, "load_all_tools", lambda: []) # nothing enabled
s = _settings()
o = offer_for_gated_server("Google Workspace", s)
assert o is not None
assert o["id"] == "Google Workspace"
assert o["id"] in VETTED
def test_offer_rejects_unvetted_and_empty(monkeypatch):
monkeypatch.setattr(pf, "load_all_tools", lambda: [])
s = _settings()
assert offer_for_gated_server("NotAVettedServer", s) is None
assert offer_for_gated_server("", s) is None
assert offer_for_gated_server(None, s) is None # type: ignore[arg-type]
def test_offer_suppressed_when_dismissed(monkeypatch):
monkeypatch.setattr(pf, "load_all_tools", lambda: [])
s = _settings({"Google Workspace": "2026-01-01T00:00:00Z"})
assert offer_for_gated_server("Google Workspace", s) is None
def test_offer_suppressed_when_already_active(monkeypatch):
monkeypatch.setattr(
pf, "load_all_tools",
lambda: [SimpleNamespace(name="Google Workspace", enabled=True)],
)
s = _settings()
assert offer_for_gated_server("Google Workspace", s) is None
def test_offer_carries_no_activate_capability(monkeypatch):
# The security invariant: an offer is data to display, never an action that grants access.
monkeypatch.setattr(pf, "load_all_tools", lambda: [])
s = _settings()
for entry in CURATED_SHORTLIST:
o = offer_for_gated_server(entry["id"], s)
assert o is not None
assert set(o.keys()) == OFFER_SHAPE, f"offer for {entry['id']} grew an unexpected field"