MCP permission model: EXTERNAL floor, durable per-tool trust, fixed approval cards (OPE-136)

coworker/agent.py

@@ -433,9 +433,13 @@ def _saving_enabled() -> bool:
         )
     )
 
-    # User-local risk overrides (mainly to relax MCP's conservative default). Empty store →
-    # no-op; never written by persona loading (the no-self-grant rule).
-    risk_overrides = RiskOverrideStore(state_dir() / "risk_overrides.json").resolver()
+    # User-local risk overrides (relax a plugin / tighten anything) + OPE-136 trust
+    # rules (per-MCP-tool "don't ask", durable). One store, never written by persona
+    # loading (the no-self-grant rule). The same instance serves the read side
+    # (classify + the trusted branch) and the write side ("Always allow this tool"),
+    # so a rule minted mid-session quiets THIS session immediately and every later
+    # one via the file.
+    override_store = RiskOverrideStore(state_dir() / "risk_overrides.json")
     permissions = PermissionEngine(
         workspace_root=ws or (root_list[0].path if root_list else Path.cwd()),
         mode=mode,
@@ -446,7 +450,9 @@ def _saving_enabled() -> bool:
         auto_allow_tools=set(config.auto_allow),
         allowed_domains=list(config.allowed_domains),
         roots=root_list or None,
-        risk_overrides=risk_overrides,
+        risk_overrides=override_store.resolver(),
+        trust_overrides=override_store.trusted,
+        grant_trust=override_store.set_trust,
     )
     # The plan-mode exit door — mutually exclusive with the board's decomposition
     # gate, DERIVED from the team trait (owner call 2026-08-16): a lead never

coworker/engine.py

@@ -47,6 +47,14 @@ class ApprovalOutcome(str, Enum):
     ALWAYS_DOMAIN = "always_domain"
     # Session-wide grant for classifier-approved read-only shell commands (readonly.py).
     READONLY_SESSION = "readonly_session"
+    # OPE-136 durable trust: persist a per-tool "don't ask" rule for an MCP tool —
+    # survives sessions, revocable on the server's detail page. MCP-only (validated
+    # server-side in manager._grant_offered, like every other grant).
+    ALWAYS_TRUST = "always_trust"
+    # OPE-136 run grant ("Allow for this request"): cover this exact tool for the
+    # remainder of the CURRENT run only — in-memory, cleared at the run boundary,
+    # nothing persisted. EXTERNAL-risk tools only (validated server-side).
+    THIS_RUN = "this_run"
     DENY = "deny"
 
 
@@ -66,6 +74,10 @@ class PermissionRequest:
     metadata: Any
     reason: str
     tool_call_id: Optional[str] = None  # for durable resume (idempotent inbox item)
+    # Where an MCP call actually goes ({transport, host}, from the server DEF at
+    # registration) — carried on the request so a PARKED approval shows the same
+    # destination evidence as the live card (§35 parity). None for non-MCP tools.
+    mcp_destination: Optional[dict] = None
 
 
 Approver = Callable[[PermissionRequest], Awaitable[ApprovalOutcome]]
@@ -312,9 +324,18 @@ async def run(
             data["source"] = source
         if display is not None:
             data["display"] = display
+        # OPE-136 run grants: a fresh run starts with a clean slate (belt — the
+        # finally below is the braces; an abandoned generator must not leak a
+        # previous answer's "Allow for this request" into this one).
+        self.permissions.clear_run_allowances()
         yield Event(EventType.TURN_START, data)
-        async for event in self._loop():
-            yield event
+        try:
+            async for event in self._loop():
+                yield event
+        finally:
+            # The run boundary IS the grant's expiry — normal finish, Stop, and
+            # generator teardown (disconnect) all land here.
+            self.permissions.clear_run_allowances()
 
     def switch_model(self, model: str) -> Optional[str]:
         """Rebind the session's model mid-conversation (roadmap item 3). History is
@@ -1147,6 +1168,33 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]"
         if allowed and decision.reason == "full access":
             self._approval_origins[tool_call.id] = {"origin": "bypass"}
 
+        # OPE-136: a trusted-MCP allow ran cardless on standing config (a user trust
+        # rule, or the legacy server flag) — audited and chip-annotated like every
+        # other cardless origin ("recorded, never invisible"). Prefix-matched against
+        # permissions.py's two trusted-branch reason strings — and the chip keeps the
+        # two apart: "your trust rule" points at the tool page's Revoke, "server
+        # trust" at the mcp.json flag. One generic label made a user believe the
+        # SERVER had marked their own rule (owner-hit 2026-08-30).
+        if allowed and decision.reason.startswith("trusted MCP tool"):
+            origin = (
+                "trusted_rule"
+                if "user trust rule" in decision.reason
+                else "trusted_server"
+            )
+            self._approval_origins[tool_call.id] = {"origin": origin}
+            self._audit(
+                tool_call, stage="auto_allowed", status="allowed", reason=reason
+            )
+
+        # OPE-136 run grant: a covered call ran cardless under the user's in-run
+        # "Allow for this request" click — silent to attention, never invisible to
+        # the record (transcript chip + audit row, like every cardless origin).
+        if allowed and decision.reason == "tool allowed for this request":
+            self._approval_origins[tool_call.id] = {"origin": "run_grant"}
+            self._audit(
+                tool_call, stage="auto_allowed", status="allowed", reason=reason
+            )
+
         if not allowed and decision.needs_user and self._consume_allow_anyway(tool_call):
             # §8.4 "Allow anyway": the human already approved this exact action from the
             # deny card. One-shot — consumed above; a different action never matches.
@@ -1263,6 +1311,21 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]"
                     # True when this shell command classifies as read-only — the card
                     # offers "Allow read-only commands for this session" only then.
                     "readonly_ok": _readonly_ok(tool_call.arguments),
+                    # OPE-136 finding 4: where an MCP call actually goes, stamped at
+                    # registration (mcp/tools.py) from the server def — so the card's
+                    # scope chip can say "leaves this computer → host" instead of the
+                    # catch-all "stays on this computer". None for non-MCP tools.
+                    **(
+                        {"mcp_destination": dest}
+                        if (
+                            dest := getattr(
+                                spec.func, "__coworker_mcp_destination__", None
+                            )
+                            if spec
+                            else None
+                        )
+                        else {}
+                    ),
                     **(
                         self.approval_extras(tool_call.name, tool_call.arguments)
                         if self.approval_extras
@@ -1284,6 +1347,11 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]"
                         metadata=metadata,
                         reason=decision.reason,
                         tool_call_id=tool_call.id,
+                        mcp_destination=(
+                            getattr(spec.func, "__coworker_mcp_destination__", None)
+                            if spec
+                            else None
+                        ),
                     )
                 ),
                 interrupted=ApprovalOutcome.DENY,
@@ -1319,6 +1387,13 @@ async def _authorize(self, tool_call: ToolCall) -> "AsyncIterator[Event | bool]"
                     )
                 elif outcome is ApprovalOutcome.READONLY_SESSION:
                     self.permissions.allow_readonly_for_session()
+                elif outcome is ApprovalOutcome.ALWAYS_TRUST:
+                    # Durable per-tool trust (OPE-136 §4): lands in the user-local
+                    # override store, so tomorrow's sessions stay quiet too.
+                    self.permissions.grant_trust_for_tool(tool_call.name)
+                elif outcome is ApprovalOutcome.THIS_RUN:
+                    # Run grant: dies with the current answer (cleared in run()).
+                    self.permissions.allow_tool_for_run(tool_call.name)
                 allowed, reason = True, "approved by user"
                 self._approval_origins[tool_call.id] = {
                     "origin": "user",

coworker/mcp/config.py

@@ -133,7 +133,10 @@ def patch_global_server(name: str, changes: dict[str, Any]) -> bool:
     servers = read_global()
     if name not in servers:
         return False
-    servers[name] = {**servers[name], **changes}
+    merged = {**servers[name], **changes}
+    # A None value DELETES the key (there is no other way to remove one through a
+    # merge patch) — used by the OPE-136 trust migration to drop `requires_approval`.
+    servers[name] = {k: v for k, v in merged.items() if v is not None}
     _write_global(servers)
     return True
 

coworker/mcp/tools.py

@@ -87,5 +87,24 @@ def _invoke(_remote: str = remote, **kwargs: Any) -> Any:
             requires_approval=server.requires_approval,
         )
         _invoke.__coworker_schema__ = _openai_schema(name, mcp_tool)
+        # OPE-136 finding 4: where this call actually goes, for the approval card's
+        # scope chip. From the server DEF (user-authored config), never from anything
+        # the server itself claims. http → the remote host; stdio → a local process.
+        _invoke.__coworker_mcp_destination__ = {
+            "transport": server.transport,
+            "host": _server_host(server),
+        }
         callables.append(_invoke)
     return callables
+
+
+def _server_host(server: MCPServerDef) -> str:
+    """The hostname an HTTP server's calls reach (lowercased), "" for stdio/unparseable."""
+    if not server.url:
+        return ""
+    try:
+        from urllib.parse import urlparse
+
+        return (urlparse(server.url).hostname or "").lower()
+    except ValueError:  # pragma: no cover - urlparse rarely raises, but fail to ""
+        return ""

coworker/overrides.py

@@ -1,8 +1,13 @@
-"""User-local risk overrides — relax (or tighten) a tool's risk class.
+"""User-local risk overrides — relax (or tighten) a tool's risk class — and, since
+OPE-136, per-tool TRUST rules.
 
-Mainly to relax MCP's conservative default (every MCP tool defaults to ``external``): a user
-who trusts a server can mark its read-only tools ``read`` so they stop gating. Rules match the
-tool name (e.g. ``mcp__notion__create_page``) by glob; the most specific rule wins.
+``rules`` relax or tighten a third-party (plugin) tool's risk class by glob; the most
+specific rule wins. MCP tools cannot be reclassified (the floor in ``risk.classify``);
+their sanctioned lever is a ``trust`` rule instead: *waive the approval card for this
+tool* — nothing else. A trusted tool stays EXTERNAL: read-only modes still deny it, the
+Auto-approve reviewer still judges it, and the audit trail still records it. One store,
+two rule types, one loader — deliberately NOT a second file (the architecture review
+rejected a parallel trust store as yet another labeling system).
 
 **Inviolable rule: this store is user-local and is NEVER written by a persona/package.** A
 persona can declare what tools it wants, but only the user decides how much to trust them — so
@@ -36,18 +41,51 @@ def _specificity(pattern: str) -> int:
 class RiskOverrideStore:
     def __init__(self, path: Optional[str | Path] = None) -> None:
         self.path = Path(path) if path else None
+        # Rules refused at load with the reason why — surfaced to the user instead of
+        # silently shaping permissions differently than their file says.
+        self.rejected: list[tuple[str, str]] = []  # (pattern, reason)
+        # OPE-136 trust rules: exact tool names (the card writes exact names — a button
+        # grants precisely what its card showed; globs stay a hand-editing power path).
+        self._trust: list[str] = []
         self._rules: list[_Rule] = self._load()
 
     def _load(self) -> list[_Rule]:
         if not (self.path and self.path.is_file()):
             return []
         data = json.loads(self.path.read_text(encoding="utf-8"))
+        # Trust entries: {"pattern": "..."} dicts (the written form) or bare strings.
+        seen: set[str] = set()
+        for entry in data.get("trust", []) or []:
+            pattern = (
+                str(entry.get("pattern", "")) if isinstance(entry, dict) else str(entry)
+            )
+            if pattern and pattern not in seen:
+                seen.add(pattern)
+                self._trust.append(pattern)
         rules = []
         for r in data.get("rules", []):
             try:
-                rules.append(_Rule(str(r["pattern"]), RiskClass(str(r["risk"]))))
+                rule = _Rule(str(r["pattern"]), RiskClass(str(r["risk"])))
             except (KeyError, ValueError):
                 continue  # skip malformed rules rather than failing the whole store
+            # OPE-136: an explicitly MCP-targeting rule may not sink a tool below
+            # EXTERNAL — the floor in risk.classify would silently ignore it anyway,
+            # and a rule that reads one way in the file but acts another is worse than
+            # a refused rule. (Generic globs that merely HAPPEN to match mcp__ names
+            # load normally; the classify floor neutralizes the loosening for those.)
+            if rule.pattern.startswith("mcp__") and rule.risk in (
+                RiskClass.READ,
+                RiskClass.EGRESS,
+            ):
+                self.rejected.append(
+                    (
+                        rule.pattern,
+                        "MCP tools cannot be reclassified below external "
+                        "(OPE-136) — use a trust rule to stop the asking",
+                    )
+                )
+                continue
+            rules.append(rule)
         return rules
 
     def save(self) -> None:
@@ -60,16 +98,26 @@ def save(self) -> None:
                     "rules": [
                         {"pattern": r.pattern, "risk": r.risk.value}
                         for r in self._rules
-                    ]
+                    ],
+                    "trust": [{"pattern": p} for p in self._trust],
                 },
                 indent=2,
             ),
             encoding="utf-8",
         )
 
     def set_rule(self, pattern: str, risk: RiskClass | str) -> None:
-        """Add/replace a user override (the everyday path writes this from the approval UI)."""
+        """Add/replace a user override (the everyday path writes this from the approval UI).
+
+        Refuses what `_load` refuses (OPE-136): an explicitly MCP-targeting rule below
+        EXTERNAL would be written now and silently dropped on the next load — a rule
+        that works for one session and then vanishes is a trap, so it never lands."""
         risk = RiskClass(risk) if not isinstance(risk, RiskClass) else risk
+        if pattern.startswith("mcp__") and risk in (RiskClass.READ, RiskClass.EGRESS):
+            raise ValueError(
+                "MCP tools cannot be reclassified below external (OPE-136) — "
+                "use a trust rule to stop the asking"
+            )
         self._rules = [r for r in self._rules if r.pattern != pattern]
         self._rules.append(_Rule(pattern, risk))
         self.save()
@@ -87,3 +135,26 @@ def resolve(self, tool_name: str) -> Optional[RiskClass]:
     def resolver(self) -> Callable[[str], Optional[RiskClass]]:
         """A callable for ``PermissionEngine.risk_overrides`` / ``risk.classify``."""
         return self.resolve
+
+    # -- OPE-136 trust rules (waive the card; never reclassify) ---------------------
+    def trusted(self, tool_name: str) -> bool:
+        """Whether a standing trust rule covers this tool (glob-matched, like risk rules)."""
+        return any(fnmatchcase(tool_name, p) for p in self._trust)
+
+    def set_trust(self, pattern: str) -> None:
+        """Mint a trust rule (the approval card's "Always allow this tool" writes an
+        EXACT name — a button grants precisely what its card showed, nothing wider)."""
+        if not pattern:
+            return
+        if pattern not in self._trust:
+            self._trust.append(pattern)
+            self.save()
+
+    def revoke_trust(self, pattern: str) -> None:
+        before = len(self._trust)
+        self._trust = [p for p in self._trust if p != pattern]
+        if len(self._trust) != before:
+            self.save()
+
+    def trust_patterns(self) -> list[str]:
+        return list(self._trust)