Fix gateway fail-direction: gate-fault on a value-moving action now fails CLOSED

Reviewer found gate_action's exception handler failed OPEN (blanket allow=True on any
error). Masked under OBSERVE today, but a live fail-open on the money path's outer
layer the moment ENFORCE is flipped. Made the fail-direction action-type-aware.

 - runtime/access_policy.py: could_move_value(action_type) — a coarse PUBLIC fail-
   direction helper. Deny-unless-clearly-a-benign-read, so a value-moving action can
   never be missed and wrongly allowed; unknown/empty labels fail closed too. Generic
   read verbs only — no thresholds, allowlists, or private action sets. The authoritative
   classification still lives in the private gate.
 - gateway/security_gate.py gate_action: on a gate fault, value-moving / unknown -> deny
   (allow=False, would_block=True, generic reason); clearly-benign reads observe-allow so
   a transient fault doesn't break them.

Part 2 sweep also fixed a PRE-EXISTING sibling fail-open (not in b575add) on the primary
gated path: ProtocolStack.pre_action's Morpheus except logged and FELL THROUGH (approved
stayed True -> tool proceeded). Now fails closed for value-moving actions via the same
coarse label (resolving the real action behind the platform_action mega-tool). Benign
reads still continue on a transient fault.

Flags unchanged (OBSERVE, App Attest enforce OFF, testnet) — this does not enable
enforcement; it makes the eventual enforcement safe. Suite 171 green; boundary clean
(generic read verbs only).

Author: Neo

gateway/security_gate.py

@@ -113,8 +113,19 @@ async def gate_action(
         gate = get_morpheus_security()
         return await gate.evaluate(action, ctx)
     except Exception:
-        logger.exception("security gate call failed; allowing (observe)")
-        return {"allow": True, "would_block": False, "route": "observe", "reason": ""}
+        # The gate is unreachable / faulted. Fail by ACTION TYPE, not blanket-allow:
+        # a value-moving (or unknown) action must NOT proceed ungated — fail CLOSED
+        # (deny) so nothing moves value when we can't reach the gate. A clearly-benign
+        # read observe-allows so a transient gateway fault doesn't break it. The coarse
+        # public label is the only input; the authoritative classification still lives
+        # in the private gate.
+        from runtime.access_policy import could_move_value
+        if could_move_value(action_type):
+            logger.exception("security gate unreachable; FAIL-CLOSED deny (action=%s)", action_type)
+            return {"allow": False, "would_block": True, "route": "fail-closed",
+                    "reason": _GENERIC_DENY}
+        logger.warning("security gate unreachable; observe-allow (benign read=%s)", action_type)
+        return {"allow": True, "would_block": False, "route": "observe-read-failopen", "reason": ""}
 
 
 def is_blocked(decision: dict) -> bool:

runtime/access_policy.py

@@ -60,6 +60,44 @@ def _is_state_modifying(action: str | None) -> bool:
         return True
 
 
+# ── Coarse fail-direction (used only when the security gate is unreachable) ──────
+#
+# Generic read verbs — the ONLY labels treated as safe to observe-allow when the
+# security gate can't be reached / faults. EVERYTHING ELSE (anything value-moving,
+# owner-gated, or simply unrecognised) FAILS CLOSED. This is a deliberately
+# conservative PUBLIC default for one decision — the fail DIRECTION — and is NOT the
+# authoritative classification (that lives in the private gate). It is intentionally
+# independent of any value-moving list so a fund-moving action can never be *missed*
+# and wrongly allowed: we allow only what is clearly a benign read, and deny the rest.
+_BENIGN_READ_LABELS = frozenset({
+    "get", "list", "read", "view", "info", "status", "health", "ping", "ready",
+    "quote", "quotes", "balance", "balances", "price", "prices", "rate", "rates",
+    "history", "feed", "search", "lookup", "metrics", "dashboard", "manifest",
+    "config", "weather", "preview", "estimate", "simulate", "positions", "portfolio",
+    "profile",
+})
+_BENIGN_READ_PREFIXES = ("get_", "list_", "read_", "fetch_", "view_", "quote_", "status_")
+
+
+def could_move_value(action_type: str | None) -> bool:
+    """Coarse PUBLIC fail-direction guess: could this action move value or change
+    security/owner state?
+
+    Used for ONE purpose only — choosing which way to fail when the security gate is
+    unreachable: ``True`` → fail CLOSED (deny), ``False`` → safe to observe-allow.
+    Anything that is not CLEARLY a benign read (incl. unknown/empty labels) is treated
+    as value-moving and fails closed. No thresholds, no allowlists, no private action
+    sets — just generic read verbs. The authoritative classification still lives in
+    the private gate; this only decides the safe direction on a gateway fault.
+    """
+    a = (action_type or "").strip().lower()
+    if not a:
+        return True  # unknown → safest direction: treat as value-moving
+    if a in _BENIGN_READ_LABELS or a.startswith(_BENIGN_READ_PREFIXES):
+        return False  # clearly a benign read → a transient fault may observe-allow it
+    return True       # value-moving / owner-gated / unrecognised → fail closed
+
+
 def default_agent_access(agent: str | None, tool: str, action: str | None = None) -> tuple[bool, str]:
     """Coarse PUBLIC per-agent access decision. Returns ``(allowed, reason)``.
 

runtime/protocols/integration.py

@@ -354,6 +354,21 @@ async def pre_action(
                     return result
             except Exception:
                 logger.exception("MorpheusSecurity pre-action failed")
+                # Fail-direction on a gate fault: a value-moving / owner-gated action
+                # must NOT proceed ungated — fail CLOSED (deny). A benign read may
+                # continue so a transient fault doesn't break it. Coarse public label
+                # only; for the platform_action mega-tool the real action is in the
+                # arguments. The authoritative classification lives in the private gate.
+                gated_label = tool_name
+                if tool_name == "platform_action" and isinstance(arguments, dict):
+                    gated_label = arguments.get("action") or tool_name
+                from runtime.access_policy import could_move_value
+                if could_move_value(gated_label):
+                    result["approved"] = False
+                    result["denial_reason"] = (
+                        "This action couldn't be authorized right now. Please try again."
+                    )
+                    return result
 
         # Rexhepi gate evaluation
         if self._rexhepi_gate is not None: