diff --git a/config/app_config.yaml b/config/app_config.yaml
index 2efcd35b..ca0ddadb 100644
--- a/config/app_config.yaml
+++ b/config/app_config.yaml
@@ -66,10 +66,27 @@ telegram:
 agent_layer:
   base_url: "http://127.0.0.1:5003"
   enabled: true
-  permission_timeout_seconds: 60
+  permission_timeout_seconds: 120
   auto_approve_user_actions: false
   trial_monthly_quota: 10
   leaky_providers:
     - openrouter
     - openai
   dev_mode_show_raw_tools: false
+
+# Two-envelope scope model (config-namespace-decision-2026-07-04.md).
+# permission: PermissionGate envelope — readable WITHOUT a permission card.
+# injection: session-start cascade envelope — auto-injected content.
+# Both share the geometry (radius, m_max, decay); injection must always be
+# a subset of permission (interactive_agent_layer/envelope.py validates this
+# at load time). Defaults below reproduce today's behavior exactly.
+scope:
+  permission:
+    radius: 3
+    m_max: 4
+    decay: 1
+  injection:
+    radius: 1
+    m_max: 2
+    decay: 0
+  scenarios: {}
diff --git a/db/pg_queries/__init__.py b/db/pg_queries/__init__.py
index 17eb1d86..6e0a7453 100644
--- a/db/pg_queries/__init__.py
+++ b/db/pg_queries/__init__.py
@@ -29,6 +29,7 @@
     rename_node, delete_node, archive_node, unarchive_node, patch_node_fields,
     get_auto_archivable_nodes, get_milestone_nodes,
     link_task_to_node, unlink_task_from_node, get_node_tasks, get_task_nodes,
+    get_node_tree_distance,
 )
 from db.pg_queries.sections import (
     get_sections, get_section, upsert_section, append_section, delete_section,
@@ -84,5 +85,5 @@
 from db.pg_queries.node_memory import (
     get_node_summary, list_node_summary_levels, upsert_node_summary,
     log_node_read, has_read_node_in_conversation, get_conversation_reads,
-    get_context_node_id_for_conversation, get_node_tree_distance,
+    get_context_node_id_for_conversation,
 )
diff --git a/db/pg_queries/node_memory.py b/db/pg_queries/node_memory.py
index cb78445b..da578fdf 100644
--- a/db/pg_queries/node_memory.py
+++ b/db/pg_queries/node_memory.py
@@ -204,43 +204,6 @@ async def get_context_node_id_for_conversation(
     return row["context_node_id"]
 
 
-async def get_node_tree_distance(
-    conn: asyncpg.Connection,
-    from_id: str,
-    to_id: str,
-    max_N: int,
-) -> int | None:
-    """Return the tree distance (edges) between two context nodes, or None if > max_N.
-
-    Uses a recursive CTE that walks both parent and child edges, bounded by max_N.
-    Returns 0 if from_id == to_id.
-    """
-    if from_id == to_id:
-        return 0
-
-    row = await conn.fetchrow(
-        """
-        WITH RECURSIVE reachable(node_id, dist) AS (
-            SELECT $1::uuid, 0
-          UNION ALL
-            SELECT
-              CASE WHEN cn.parent_id = r.node_id THEN cn.id
-                   ELSE cn.parent_id
-              END,
-              r.dist + 1
-            FROM reachable r
-            JOIN context_nodes cn
-              ON (cn.id = r.node_id AND cn.parent_id IS NOT NULL)
-              OR (cn.parent_id = r.node_id)
-            WHERE r.dist < $3
-        )
-        SELECT dist FROM reachable WHERE node_id = $2::uuid LIMIT 1
-        """,
-        _uuid.UUID(from_id), _uuid.UUID(to_id), max_N,
-    )
-    return row["dist"] if row else None
-
-
 async def get_conversation_reads(
     conn: asyncpg.Connection,
     conversation_id: str,
diff --git a/db/pg_queries/nodes.py b/db/pg_queries/nodes.py
index 6cfe29d7..e06504b6 100644
--- a/db/pg_queries/nodes.py
+++ b/db/pg_queries/nodes.py
@@ -411,59 +411,59 @@ async def get_milestone_nodes(
     return [_node(r) for r in rows]
 
 
-# ── Hop distance ─────────────────────────────────────────────────────────────
-
-
-async def get_node_hop_distance(
+# ── Tree distance (bounded BFS, multi-root) ────────────────────────────────
+#
+# This is the SOLE distance query — consolidated from two prior
+# implementations: this module's now-deleted unbounded LCA-walk
+# get_node_hop_distance, and node_memory.get_node_tree_distance (bounded
+# single-root BFS, moved here and generalized to multi-root). The gate
+# (PermissionGate) always has a radius to bound the walk by, so LCA's
+# unbounded ancestor-chain cost is unnecessary; bounded BFS also lets the
+# caller cap query cost independent of tree depth.
+
+
+async def get_node_tree_distance(
     conn: asyncpg.Connection,
-    from_node_id: str,
-    to_node_id: str,
+    from_ids: list[str],
+    to_id: str,
+    max_N: int,
 ) -> int | None:
-    """Return the undirected tree distance (hop count) between two context nodes.
-
-    Uses an LCA (lowest common ancestor) approach: walk ancestor chains from
-    both nodes and find the shortest combined path via their common ancestor.
+    """Return the minimum tree distance (edges) from any of *from_ids* to *to_id*.
 
-    Returns:
-        int  — number of hops on the shortest tree path.
-        None — nodes are in separate trees (no common ancestor).
+    Multi-root seedable: all *from_ids* are seeded at distance 0 and the
+    minimum distance across all roots wins (DD §5.8's multi-root org-accounts
+    model). Single-root callers pass ``[id]``.
 
-    Performance note: for large trees this CTE scans O(depth) rows per node.
-    A materialized path column or a dedicated ancestor table would reduce this
-    to O(1) lookups at the cost of write overhead. Consider adding
-    ``ltree`` indexing or a closure table if this becomes a hot path.
+    Uses a recursive CTE that walks both parent and child edges, bounded by
+    max_N. Returns None if the true distance exceeds max_N (or no path
+    exists within the bound).
     """
-    result = await conn.fetchval(
+    if to_id in from_ids:
+        return 0
+
+    row = await conn.fetchrow(
         """
-        WITH RECURSIVE
-          from_ancestors(id, depth) AS (
-            SELECT id, 0 AS depth
-            FROM context_nodes
-            WHERE id = $1
-            UNION ALL
-            SELECT cn.parent_id, fa.depth + 1
-            FROM context_nodes cn
-            JOIN from_ancestors fa ON cn.id = fa.id
-            WHERE cn.parent_id IS NOT NULL
-          ),
-          to_ancestors(id, depth) AS (
-            SELECT id, 0 AS depth
-            FROM context_nodes
-            WHERE id = $2
-            UNION ALL
-            SELECT cn.parent_id, ta.depth + 1
-            FROM context_nodes cn
-            JOIN to_ancestors ta ON cn.id = ta.id
-            WHERE cn.parent_id IS NOT NULL
-          )
-        SELECT MIN(fa.depth + ta.depth)
-        FROM from_ancestors fa
-        JOIN to_ancestors ta ON fa.id = ta.id
+        WITH RECURSIVE reachable(node_id, dist) AS (
+            SELECT unnest($1::uuid[]), 0
+          UNION ALL
+            SELECT
+              CASE WHEN cn.parent_id = r.node_id THEN cn.id
+                   ELSE cn.parent_id
+              END,
+              r.dist + 1
+            FROM reachable r
+            JOIN context_nodes cn
+              ON (cn.id = r.node_id AND cn.parent_id IS NOT NULL)
+              OR (cn.parent_id = r.node_id)
+            WHERE r.dist < $3
+        )
+        SELECT MIN(dist) AS dist FROM reachable WHERE node_id = $2::uuid
         """,
-        _uuid.UUID(from_node_id),
-        _uuid.UUID(to_node_id),
+        [_uuid.UUID(fid) for fid in from_ids],
+        _uuid.UUID(to_id),
+        max_N,
     )
-    return int(result) if result is not None else None
+    return row["dist"] if row and row["dist"] is not None else None
 
 
 # ── Node-task linking ─────────────────────────────────────────────────────────
diff --git a/interactive_agent_layer/config.py b/interactive_agent_layer/config.py
index d6dc1a19..1cb0e7da 100644
--- a/interactive_agent_layer/config.py
+++ b/interactive_agent_layer/config.py
@@ -13,7 +13,7 @@ def is_enabled() -> bool:
 
 
 def get_permission_timeout() -> int:
-    return int(config.get("agent_layer.permission_timeout_seconds", 900))
+    return int(config.get("agent_layer.permission_timeout_seconds", 120))
 
 
 def get_auto_approve_user_actions() -> bool:
@@ -28,13 +28,3 @@ def get_trial_monthly_quota() -> int:
 def get_leaky_providers() -> list[str]:
     """Providers whose dashboards expose prompt content — 2.5 is disabled on these."""
     return list(config.get("agent_layer.leaky_providers", ["openrouter", "openai"]))
-
-
-def get_scope_radius() -> int:
-    """Default M (hop radius) for read_context scope gating.
-
-    Used when a session doesn't set scope_radius explicitly in options —
-    e.g. when scope_source_node_id is resolved from the conversation's
-    context_node_id at session-start rather than passed in by the caller.
-    """
-    return int(config.get("agent_layer.scope_radius", 3))
diff --git a/interactive_agent_layer/envelope.py b/interactive_agent_layer/envelope.py
new file mode 100644
index 00000000..b46e1433
--- /dev/null
+++ b/interactive_agent_layer/envelope.py
@@ -0,0 +1,133 @@
+"""Two-envelope scope model: PermissionGate and injection-cascade envelopes.
+
+Both envelopes share one geometry `(radius, m_max, decay)` under the top-level
+`scope:` config namespace (config-namespace-decision-2026-07-04.md):
+
+    scope:
+      permission:   # PermissionGate envelope — readable WITHOUT a permission card
+        radius: 3
+        m_max: 4
+        decay: 1
+      injection:    # session-start cascade envelope — auto-injected content
+        radius: 1
+        m_max: 2
+        decay: 0
+      scenarios: {}  # per-scenario partial overrides, e.g. scenarios.<name>.permission
+
+`load_injection_envelope()` always validates the injection envelope is a
+subset of the permission envelope for the same scenario (`validate_injection_subset`)
+and raises `ScopeConfigError` at load time on violation — fail fast, no clamping.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from config.loader import config
+
+# Defaults preserve today's behavior: permission 3/4/1 matches the existing
+# agent_layer.scope_radius default (3) with the D04 4/3/2/1 ring curve;
+# injection 1/2/0 matches the current flat cascade (depth 1, M=2).
+_PERMISSION_DEFAULT = {"radius": 3, "m_max": 4, "decay": 1}
+_INJECTION_DEFAULT = {"radius": 1, "m_max": 2, "decay": 0}
+
+# The distance-query call sites (db.pg_queries.nodes.get_node_tree_distance,
+# and premium's scope_grants.make_hop_distance_fn) bound their bounded-BFS
+# walk to this same constant as max_N. A configured permission.radius beyond
+# this bound would never actually be honored by the DB query — it would
+# silently behave as if capped rather than enforcing the configured radius.
+# load_permission_envelope() rejects that at load time instead (fail fast,
+# no silent under-enforcement, consistent with validate_injection_subset's
+# raise-not-clamp policy).
+MAX_HOP_DISTANCE_BOUND = 20
+
+
+class ScopeConfigError(Exception):
+    """Raised when the injection envelope is not a subset of the permission
+    envelope for the same scenario — fail fast at config-load time."""
+
+
+@dataclass(frozen=True)
+class ScopeEnvelope:
+    """A graded scope geometry: how far (radius) and how much detail
+    (m_max, decayed per hop) is allowed at each tree distance."""
+
+    radius: int
+    m_max: int
+    decay: int
+
+    def m_allowed(self, d: int) -> int | None:
+        """Detail tier allowed at tree distance *d*, or None if *d* is out of
+        scope entirely (d > radius). Clamped to a minimum of 1 within radius."""
+        if d > self.radius:
+            return None
+        return max(1, self.m_max - self.decay * d)
+
+
+def _merge_scenario(base: dict, scenario: str | None, subkey: str) -> dict:
+    """Partial-merge a per-scenario override onto the base envelope dict."""
+    if scenario is None:
+        return base
+    override = config.get(f"scope.scenarios.{scenario}.{subkey}", {}) or {}
+    return {**base, **override}
+
+
+def load_permission_envelope(scenario: str | None = None) -> ScopeEnvelope:
+    """Load the PermissionGate envelope from `scope.permission`, with an
+    optional per-scenario partial override from `scope.scenarios.<name>.permission`.
+
+    Raises ScopeConfigError if radius exceeds MAX_HOP_DISTANCE_BOUND — the
+    distance query bounds its walk to that same constant, so a larger
+    configured radius would never be honored.
+    """
+    base = config.get("scope.permission", _PERMISSION_DEFAULT) or _PERMISSION_DEFAULT
+    merged = _merge_scenario(base, scenario, "permission")
+    envelope = ScopeEnvelope(**merged)
+    if envelope.radius > MAX_HOP_DISTANCE_BOUND:
+        raise ScopeConfigError(
+            f"permission envelope radius ({envelope.radius}) exceeds "
+            f"MAX_HOP_DISTANCE_BOUND ({MAX_HOP_DISTANCE_BOUND}) — the distance "
+            f"query would never honor a radius this large (scope.permission"
+            f"{f'.scenarios.{scenario}' if scenario else ''})"
+        )
+    return envelope
+
+
+def load_injection_envelope(scenario: str | None = None) -> ScopeEnvelope:
+    """Load the injection-cascade envelope from `scope.injection`, with an
+    optional per-scenario partial override from `scope.scenarios.<name>.injection`.
+
+    Validates the loaded envelope is a subset of the same-scenario permission
+    envelope on every call — raises ScopeConfigError on violation.
+    """
+    base = config.get("scope.injection", _INJECTION_DEFAULT) or _INJECTION_DEFAULT
+    merged = _merge_scenario(base, scenario, "injection")
+    injection = ScopeEnvelope(**merged)
+    permission = load_permission_envelope(scenario)
+    validate_injection_subset(injection, permission)
+    return injection
+
+
+def validate_injection_subset(inj: ScopeEnvelope, perm: ScopeEnvelope) -> None:
+    """Enforce `injection ⊆ permission`: inj.radius <= perm.radius AND
+    inj.m_allowed(d) <= perm.m_allowed(d) for every d in [0, inj.radius].
+
+    The full loop is required, not just endpoints — two clamped-linear curves
+    can cross mid-range. Raises ScopeConfigError naming both envelopes on
+    the first violating distance found.
+    """
+    if inj.radius > perm.radius:
+        raise ScopeConfigError(
+            f"injection envelope radius ({inj.radius}) exceeds permission "
+            f"envelope radius ({perm.radius}) — injection must be a subset "
+            f"of permission (scope.injection vs scope.permission)"
+        )
+    for d in range(0, inj.radius + 1):
+        inj_m = inj.m_allowed(d)
+        perm_m = perm.m_allowed(d)
+        if inj_m is not None and (perm_m is None or inj_m > perm_m):
+            raise ScopeConfigError(
+                f"injection envelope allows more detail than permission "
+                f"envelope at distance {d} (injection m_allowed={inj_m}, "
+                f"permission m_allowed={perm_m}) — injection must be a "
+                f"subset of permission (scope.injection vs scope.permission)"
+            )
diff --git a/interactive_agent_layer/permissions.py b/interactive_agent_layer/permissions.py
index ef793ff6..76ffd1b3 100644
--- a/interactive_agent_layer/permissions.py
+++ b/interactive_agent_layer/permissions.py
@@ -5,9 +5,10 @@
 import dataclasses
 import uuid
 from collections.abc import Callable, Awaitable
-from typing import Any
+from typing import Any, Literal
 
 from interactive_agent_layer.config import get_permission_timeout
+from interactive_agent_layer.envelope import ScopeEnvelope
 from interactive_agent_layer.translation import (
     BackgroundEntry,
     BackgroundHiddenEntry,
@@ -49,6 +50,8 @@ def __init__(self, request_id: str) -> None:
 # path -> node_id | None
 ResolveNodePathFn = Callable[[str], Awaitable[str | None]]
 
+OnTimeout = Literal["deny", "raise"]
+
 
 class PermissionGate:
     """
@@ -58,8 +61,10 @@ class PermissionGate:
         async (tool_name: str, args: dict, ctx: Any) -> PermissionResultAllow | PermissionResultDeny
 
     ``outbound_events`` receives permission_request dicts when a user_action tool
-    needs approval.  Callers (run_turn) drain this queue and yield the events on
-    the SSE stream so they cross the process boundary to the API / dispatch layer.
+    needs approval, and permission_resolved dicts on every terminal resolution
+    (approved/denied/timeout). Callers (run_turn) drain this queue and yield the
+    events on the SSE stream so they cross the process boundary to the API /
+    dispatch layer.
 
     Grant functions (optional, injected for testability and DB-decoupling):
         check_grant_fn(user_id, conversation_id, target, kind) -> bool
@@ -67,13 +72,16 @@ class PermissionGate:
             auto-allows and skips the interactive permission_request flow.
         insert_grant_fn(user_id, conversation_id, target, kind) -> None
             Called after the user approves to persist the grant for the
-            remainder of the conversation.
+            remainder of the conversation. Applies to BOTH the scope-read path
+            (kind="read_out_of_scope") and the user_action translation-table
+            path — grants are checked/inserted uniformly on both.
 
     Scope-gating callables (optional, injected; None = no scope enforcement):
         scope_source_node_id: anchor node for this conversation's scope.
-        scope_radius: M hops from source — targets beyond this distance are gated.
-        scope_mode: "tree_distance" (undirected shortest path, default) or
-            "descendant_only" (reserved for future use; treated as tree_distance).
+        scope_envelope: ScopeEnvelope — graded (radius, m_max, decay) geometry.
+            A read_context target is out of scope iff its tree distance d from
+            scope_source_node_id exceeds scope_envelope.radius, OR the tool
+            call's requested M exceeds scope_envelope.m_allowed(d).
         hop_distance_fn(from_id, to_id) -> int | None
             Returns undirected tree distance; None means unrelated trees.
             Pre-bound to a DB connection in production wiring; mocked in tests.
@@ -93,8 +101,7 @@ def __init__(
         insert_grant_fn: InsertGrantFn | None = None,
         # Scope-gating params — all None means no scope enforcement
         scope_source_node_id: str | None = None,
-        scope_radius: int | None = None,
-        scope_mode: str = "tree_distance",
+        scope_envelope: ScopeEnvelope | None = None,
         hop_distance_fn: HopDistanceFn | None = None,
         resolve_node_path_fn: ResolveNodePathFn | None = None,
     ) -> None:
@@ -105,8 +112,7 @@ def __init__(
         self._check_grant_fn = check_grant_fn
         self._insert_grant_fn = insert_grant_fn
         self._scope_source_node_id = scope_source_node_id
-        self._scope_radius = scope_radius
-        self._scope_mode = scope_mode
+        self._scope_envelope = scope_envelope
         self._hop_distance_fn = hop_distance_fn
         self._resolve_node_path_fn = resolve_node_path_fn
 
@@ -119,11 +125,11 @@ async def can_use_tool(
         # Scope check for read_context runs before normal translation dispatch.
         # read_context is classified as "background" (auto-allow) in the translation
         # table, but must be gated when its targets exceed the conversation's scope
-        # radius.  All three conditions must be set; if any is None, no gating.
+        # envelope. All three conditions must be set; if any is None, no gating.
         if (
             tool_name == "read_context"
             and self._scope_source_node_id is not None
-            and self._scope_radius is not None
+            and self._scope_envelope is not None
             and self._hop_distance_fn is not None
         ):
             return await self._check_read_context_scope(args)
@@ -135,13 +141,18 @@ async def _check_read_context_scope(
         self,
         args: dict[str, Any],
     ) -> PermissionResultAllow | PermissionResultDeny:
-        """Check whether read_context targets are within the session's scope radius.
+        """Check whether read_context targets are within the session's scope envelope.
 
         Collects all target node_ids (from node_ids arg and resolved paths),
-        checks each against the hop distance, and gates on the first offender.
-        An unresolvable path (resolver returns None or no resolver provided)
-        is treated as out-of-scope to prevent bypass via path arguments.
+        checks each against the graded envelope (d > radius OR requested_M >
+        m_allowed(d)), and gates on the first offender. An unresolvable path
+        (resolver returns None or no resolver provided) is treated as
+        out-of-scope to prevent bypass via path arguments.
         """
+        envelope = self._scope_envelope
+        assert envelope is not None  # narrowed by can_use_tool's guard
+        requested_M = args.get("M", 4)
+
         node_ids: list[str] = list(args.get("node_ids") or [])
         paths: list[str] = list(args.get("paths") or [])
 
@@ -167,7 +178,8 @@ async def _check_read_context_scope(
                 offender_label = "(unresolved path)"
                 break
             distance = await self._hop_distance_fn(self._scope_source_node_id, nid)
-            if distance is None or distance > self._scope_radius:
+            m_allowed = envelope.m_allowed(distance) if distance is not None else None
+            if distance is None or m_allowed is None or requested_M > m_allowed:
                 offender = nid
                 offender_label = nid
                 break
@@ -175,42 +187,31 @@ async def _check_read_context_scope(
             # All targets within scope
             return PermissionResultAllow()
 
-        # Gate: emit permission_request and await user response
-        return await self._emit_scope_permission_request(offender_label)
-
-    async def _emit_scope_permission_request(
-        self,
-        target_label: str,
-    ) -> PermissionResultAllow | PermissionResultDeny:
-        """Emit a read_out_of_scope permission_request and await user decision."""
-        request_id = str(uuid.uuid4())
+        conv_id: str = getattr(self._session, "conversation_id", None) or ""
+        kind = "read_out_of_scope"
 
-        loop = asyncio.get_running_loop()
-        future: asyncio.Future[bool] = loop.create_future()
-        self._session.permission_pending[request_id] = future
+        # Check existing per-conversation grant — skip the interactive flow if found.
+        # (DD 3.2: the scope-read path previously lacked this check entirely.)
+        if self._check_grant_fn is not None:
+            granted = await self._check_grant_fn(
+                self._session.user_id, conv_id, offender_label, kind
+            )
+            if granted:
+                return PermissionResultAllow()
 
-        await self._outbound_events.put(
-            {
-                "type": "permission_request",
-                "kind": "read_out_of_scope",
-                "session_id": self._session.session_id,
-                "request_id": request_id,
-                "target": target_label,
-                "reason_from_bot": None,
-            }
+        result = await self._await_user_decision(
+            kind=kind,
+            target=offender_label,
+            on_timeout="deny",
+            reason_from_bot=args.get("reason"),
         )
 
-        try:
-            approved = await asyncio.wait_for(
-                asyncio.shield(future),
-                timeout=get_permission_timeout(),
+        if isinstance(result, PermissionResultAllow) and self._insert_grant_fn is not None:
+            await self._insert_grant_fn(
+                self._session.user_id, conv_id, offender_label, kind
             )
-        except asyncio.TimeoutError:
-            approved = False
-        finally:
-            self._session.permission_pending.pop(request_id, None)
 
-        return PermissionResultAllow() if approved else PermissionResultDeny()
+        return result
 
     async def _dispatch(
         self,
@@ -238,10 +239,46 @@ async def _dispatch(
             if granted:
                 return PermissionResultAllow()
 
-        # Emit permission_request on the outbound queue — run_turn drains this and
-        # yields the event on the SSE stream so it crosses the process boundary to
-        # dispatch / the API layer.  WSPublisher is intentionally not used here: it
-        # only works within the same OS process.
+        # Writes (user_section_edit / destructive) raise PermissionTimeoutError
+        # on timeout — session.py catches it, emits session_timeout, and ends
+        # the turn cleanly (DD §4.5).
+        result = await self._await_user_decision(
+            kind=kind,
+            target=target,
+            on_timeout="raise",
+            reason_from_bot=args.get("reason"),
+        )
+
+        if isinstance(result, PermissionResultAllow) and self._insert_grant_fn is not None:
+            await self._insert_grant_fn(
+                self._session.user_id, conv_id, target, kind
+            )
+
+        return result
+
+    async def _await_user_decision(
+        self,
+        kind: str,
+        target: str,
+        *,
+        on_timeout: OnTimeout,
+        reason_from_bot: str | None = None,
+    ) -> PermissionResultAllow | PermissionResultDeny:
+        """Emit permission_request, await the user's decision, emit permission_resolved.
+
+        Single implementation shared by the scope-read path and the user_action
+        translation-table path (DD §5.7's "one _await_user_decision helper").
+
+        on_timeout="deny": timeout resolves as denied, no exception raised
+            (reads: deny-and-continue, DD §4.5).
+        on_timeout="raise": timeout raises PermissionTimeoutError after emitting
+            permission_resolved — session.py catches it and emits session_timeout
+            (writes: raise -> session_timeout, DD §4.5).
+
+        Emits permission_resolved on every terminal resolution (approved/denied/
+        timeout), regardless of on_timeout — this is unconditional per DD §5.7's
+        "one permission transport" requirement.
+        """
         request_id = str(uuid.uuid4())
 
         loop = asyncio.get_running_loop()
@@ -255,25 +292,32 @@ async def _dispatch(
                 "request_id": request_id,
                 "kind": kind,
                 "target": target,
-                "reason_from_bot": None,
+                "reason_from_bot": reason_from_bot,
             }
         )
 
+        timed_out = False
         try:
             approved = await asyncio.wait_for(
                 asyncio.shield(future),
                 timeout=get_permission_timeout(),
             )
         except asyncio.TimeoutError:
-            # Raise instead of silently denying — session.py catches this,
-            # emits a session_timeout event, and ends the turn cleanly.
-            raise PermissionTimeoutError(request_id)
+            timed_out = True
+            approved = False
         finally:
             self._session.permission_pending.pop(request_id, None)
 
-        if approved and self._insert_grant_fn is not None:
-            await self._insert_grant_fn(
-                self._session.user_id, conv_id, target, kind
-            )
+        resolution = "timeout" if timed_out else ("approved" if approved else "denied")
+        await self._outbound_events.put(
+            {
+                "type": "permission_resolved",
+                "request_id": request_id,
+                "resolution": resolution,
+            }
+        )
+
+        if timed_out and on_timeout == "raise":
+            raise PermissionTimeoutError(request_id)
 
         return PermissionResultAllow() if approved else PermissionResultDeny()
diff --git a/interactive_agent_layer/session.py b/interactive_agent_layer/session.py
index c8fe25d9..26866309 100644
--- a/interactive_agent_layer/session.py
+++ b/interactive_agent_layer/session.py
@@ -13,7 +13,8 @@
 from typing import AsyncIterator, Any
 
 from interactive_agent_layer.coalescing import CoalescingBuffer
-from interactive_agent_layer.config import get_auto_approve_user_actions, get_scope_radius
+from interactive_agent_layer.config import get_auto_approve_user_actions
+from interactive_agent_layer.envelope import ScopeEnvelope, load_permission_envelope
 from interactive_agent_layer.permissions import (
     CheckGrantFn,
     InsertGrantFn,
@@ -65,7 +66,7 @@ class Session:
     # options_hash for subprocess partitioning, and mutating it after the hash
     # is computed on turn 1 would silently repartition on turn 2+.
     scope_source_node_id: str | None = None
-    scope_radius: int | None = None
+    scope_envelope: ScopeEnvelope | None = None
     _scope_resolved: bool = False
 
 
@@ -183,7 +184,7 @@ async def session_resolve_node_path_fn(path: str) -> str | None:
             check_grant_fn=self.check_grant_fn,
             insert_grant_fn=self.insert_grant_fn,
             scope_source_node_id=session.scope_source_node_id,
-            scope_radius=session.scope_radius,
+            scope_envelope=session.scope_envelope,
             hop_distance_fn=session_hop_distance_fn,
             resolve_node_path_fn=session_resolve_node_path_fn,
         )
@@ -249,7 +250,7 @@ async def interrupt(self, session_id: str) -> None:
             await self.pool_client.interrupt(handle)
 
     async def _resolve_scope(self, session: Session) -> None:
-        """Resolve session.scope_source_node_id / scope_radius, once per session.
+        """Resolve session.scope_source_node_id / scope_envelope, once per session.
 
         Precedence for scope_source_node_id:
           1. session.options["scope_source_node_id"] (explicit caller override)
@@ -257,10 +258,17 @@ async def _resolve_scope(self, session: Session) -> None:
              the conversation's context_node_id (premium-injected; None if
              absent, no conversation_id, or the conversation has no node)
 
-        scope_radius: session.options["scope_radius"] if set, else the
-        config-driven default (get_scope_radius()) — but only when a
-        scope_source_node_id was actually resolved, so an unscoped session
-        never gets a radius with no anchor to measure from.
+        scope_envelope: session.options["permission_envelope"] if set (a
+        whole-envelope override, caller-supplied at session spawn — NEVER
+        model-supplied), else the config-driven default
+        (load_permission_envelope()) — but only when a scope_source_node_id
+        was actually resolved, so an unscoped session never gets an envelope
+        with no anchor to measure from.
+
+        The option value must be a plain dict (radius/m_max/decay), not a
+        ScopeEnvelope instance: session.options feeds _stable_options_hash's
+        json.dumps for the pool's subprocess partitioning, and ScopeEnvelope
+        (a dataclass) is not JSON-serializable.
 
         Cached via session._scope_resolved so this DB-touching resolution
         runs at most once per session, not once per turn.
@@ -278,12 +286,16 @@ async def _resolve_scope(self, session: Session) -> None:
                 session.user_id, session.conversation_id
             )
 
-        scope_radius = session.options.get("scope_radius")
-        if scope_radius is None and scope_source_node_id is not None:
-            scope_radius = get_scope_radius()
+        scope_envelope_option = session.options.get("permission_envelope")
+        if scope_envelope_option is not None:
+            scope_envelope = ScopeEnvelope(**scope_envelope_option)
+        elif scope_source_node_id is not None:
+            scope_envelope = load_permission_envelope()
+        else:
+            scope_envelope = None
 
         session.scope_source_node_id = scope_source_node_id
-        session.scope_radius = scope_radius
+        session.scope_envelope = scope_envelope
         session._scope_resolved = True
 
 
diff --git a/tests/db/test_node_hop_distance.py b/tests/db/test_node_hop_distance.py
deleted file mode 100644
index ca34a592..00000000
--- a/tests/db/test_node_hop_distance.py
+++ /dev/null
@@ -1,114 +0,0 @@
-"""Tests for get_node_hop_distance — uses mocked asyncpg connection.
-
-All tests run without a live Postgres instance (no DATABASE_URL required).
-The SQL logic is verified end-to-end in test_pg_nodes.py (requires DATABASE_URL).
-"""
-from __future__ import annotations
-
-import decimal
-import uuid
-from unittest.mock import AsyncMock, MagicMock
-
-import pytest
-
-from db.pg_queries.nodes import get_node_hop_distance
-
-
-# Stable test UUIDs — use fixed values so failures are easy to read
-_A = "00000000-0000-0000-0000-000000000001"
-_B = "00000000-0000-0000-0000-000000000002"
-_C = "00000000-0000-0000-0000-000000000003"
-_D = "00000000-0000-0000-0000-000000000004"
-
-
-def _mock_conn(fetchval_return):
-    """Return a mock asyncpg Connection that returns a preset value from fetchval."""
-    conn = MagicMock()
-    conn.fetchval = AsyncMock(return_value=fetchval_return)
-    return conn
-
-
-# ---------------------------------------------------------------------------
-# Basic distance cases
-# ---------------------------------------------------------------------------
-
-
-async def test_same_node_returns_zero():
-    """A node is 0 hops from itself (LCA is itself, both depths=0)."""
-    conn = _mock_conn(0)
-    result = await get_node_hop_distance(conn, _A, _A)
-    assert result == 0
-
-
-async def test_direct_parent_child_returns_one():
-    """Parent → child is 1 hop."""
-    conn = _mock_conn(1)
-    result = await get_node_hop_distance(conn, _A, _B)
-    assert result == 1
-
-
-async def test_grandparent_to_grandchild_returns_two():
-    """Grandparent → grandchild is 2 hops."""
-    conn = _mock_conn(2)
-    result = await get_node_hop_distance(conn, _A, _C)
-    assert result == 2
-
-
-async def test_sibling_returns_two():
-    """Two siblings share a common parent → 2 undirected hops (via parent)."""
-    conn = _mock_conn(2)
-    result = await get_node_hop_distance(conn, _B, _C)
-    assert result == 2
-
-
-async def test_cousins_return_four():
-    """Cousins: each 2 hops from their common grandparent → 4 hops total."""
-    conn = _mock_conn(4)
-    result = await get_node_hop_distance(conn, _C, _D)
-    assert result == 4
-
-
-# ---------------------------------------------------------------------------
-# Unrelated nodes
-# ---------------------------------------------------------------------------
-
-
-async def test_unrelated_nodes_returns_none():
-    """Nodes in separate trees (no common ancestor) → None."""
-    conn = _mock_conn(None)
-    result = await get_node_hop_distance(conn, _A, _D)
-    assert result is None
-
-
-# ---------------------------------------------------------------------------
-# Return type coercion
-# ---------------------------------------------------------------------------
-
-
-async def test_returns_int_not_decimal():
-    """Result from DB (may be Decimal/numeric) is coerced to Python int."""
-    conn = _mock_conn(decimal.Decimal("3"))
-    result = await get_node_hop_distance(conn, _A, _B)
-    assert result == 3
-    assert isinstance(result, int)
-
-
-# ---------------------------------------------------------------------------
-# Query contract — both UUIDs passed to fetchval
-# ---------------------------------------------------------------------------
-
-
-async def test_fetchval_called_with_both_uuid_args():
-    """The function passes both node IDs as uuid.UUID objects to fetchval."""
-    conn = _mock_conn(1)
-    from_id = str(uuid.uuid4())
-    to_id = str(uuid.uuid4())
-
-    await get_node_hop_distance(conn, from_id, to_id)
-
-    conn.fetchval.assert_called_once()
-    call_args = conn.fetchval.call_args
-    # positional args after the SQL string should include the two UUIDs
-    positional = call_args.args if call_args.args else call_args[0]
-    assert uuid.UUID(from_id) in positional
-    assert uuid.UUID(to_id) in positional
diff --git a/tests/db/test_node_tree_distance.py b/tests/db/test_node_tree_distance.py
new file mode 100644
index 00000000..0906b43d
--- /dev/null
+++ b/tests/db/test_node_tree_distance.py
@@ -0,0 +1,110 @@
+"""Tests for get_node_tree_distance — the consolidated bounded-BFS distance query.
+
+Uses mocked asyncpg connections (no live Postgres required). This is the SOLE
+distance query, consolidated from two prior implementations: the deleted
+unbounded LCA-walk get_node_hop_distance (nodes.py) and the deleted
+single-root bounded-BFS get_node_tree_distance (node_memory.py) — this
+version lives in nodes.py, generalized to multi-root (from_ids: list[str],
+min-distance-wins per DD §5.8's multi-root org-accounts model).
+
+The SQL logic itself is verified end-to-end in test_pg_nodes.py (requires
+DATABASE_URL); these tests verify the Python-level contract: the from_id==to_id
+short-circuit, multi-root membership short-circuit, return-value passthrough,
+and that both UUIDs/bound are threaded to the query correctly.
+"""
+from __future__ import annotations
+
+import uuid
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from db.pg_queries.nodes import get_node_tree_distance
+
+
+_A = "00000000-0000-0000-0000-000000000001"
+_B = "00000000-0000-0000-0000-000000000002"
+_C = "00000000-0000-0000-0000-000000000003"
+
+
+def _mock_conn(fetchrow_return: dict | None):
+    conn = MagicMock()
+    conn.fetchrow = AsyncMock(return_value=fetchrow_return)
+    return conn
+
+
+# ---------------------------------------------------------------------------
+# Short-circuits — no DB round trip
+# ---------------------------------------------------------------------------
+
+
+async def test_to_id_equals_only_from_id_returns_zero_without_query():
+    conn = _mock_conn(None)
+    result = await get_node_tree_distance(conn, [_A], _A, max_N=5)
+    assert result == 0
+    conn.fetchrow.assert_not_called()
+
+
+async def test_to_id_is_one_of_multiple_from_ids_returns_zero_without_query():
+    """Multi-root: to_id matching ANY root is distance 0, no query needed."""
+    conn = _mock_conn(None)
+    result = await get_node_tree_distance(conn, [_A, _B], _B, max_N=5)
+    assert result == 0
+    conn.fetchrow.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Query path — value passthrough
+# ---------------------------------------------------------------------------
+
+
+async def test_returns_distance_from_query_row():
+    conn = _mock_conn({"dist": 2})
+    result = await get_node_tree_distance(conn, [_A], _C, max_N=5)
+    assert result == 2
+
+
+async def test_returns_none_when_no_row():
+    """No row (empty CTE result) → out of bound / unreachable."""
+    conn = _mock_conn(None)
+    result = await get_node_tree_distance(conn, [_A], _C, max_N=5)
+    assert result is None
+
+
+async def test_returns_none_when_dist_is_none():
+    """Row exists but dist is NULL (e.g. MIN() over empty set) → None."""
+    conn = _mock_conn({"dist": None})
+    result = await get_node_tree_distance(conn, [_A], _C, max_N=5)
+    assert result is None
+
+
+# ---------------------------------------------------------------------------
+# Query contract — args threaded correctly
+# ---------------------------------------------------------------------------
+
+
+async def test_fetchrow_called_with_uuid_list_and_bound():
+    conn = _mock_conn({"dist": 1})
+    from_id = str(uuid.uuid4())
+    to_id = str(uuid.uuid4())
+
+    await get_node_tree_distance(conn, [from_id], to_id, max_N=7)
+
+    conn.fetchrow.assert_called_once()
+    call_args = conn.fetchrow.call_args
+    positional = call_args.args if call_args.args else call_args[0]
+    assert positional[1] == [uuid.UUID(from_id)]
+    assert positional[2] == uuid.UUID(to_id)
+    assert positional[3] == 7
+
+
+async def test_multi_root_from_ids_all_passed_as_uuid_list():
+    conn = _mock_conn({"dist": 3})
+    a, b, c = str(uuid.uuid4()), str(uuid.uuid4()), str(uuid.uuid4())
+    to_id = str(uuid.uuid4())
+
+    await get_node_tree_distance(conn, [a, b, c], to_id, max_N=5)
+
+    call_args = conn.fetchrow.call_args
+    positional = call_args.args if call_args.args else call_args[0]
+    assert positional[1] == [uuid.UUID(a), uuid.UUID(b), uuid.UUID(c)]
diff --git a/tests/interactive_agent_layer/test_config.py b/tests/interactive_agent_layer/test_config.py
index 8bc1c1f0..738c45ba 100644
--- a/tests/interactive_agent_layer/test_config.py
+++ b/tests/interactive_agent_layer/test_config.py
@@ -4,21 +4,30 @@
 from interactive_agent_layer import config as layer_config
 
 
-def test_get_scope_radius_default(monkeypatch):
-    """Defaults to a sane radius when agent_layer.scope_radius is unset."""
+def test_get_permission_timeout_default(monkeypatch):
+    """Defaults to 120s when agent_layer.permission_timeout_seconds is unset.
+
+    get_scope_radius() was deleted (replaced by envelope.load_permission_envelope());
+    scope radius now lives in the ScopeEnvelope, not a bare config int.
+    """
     monkeypatch.setattr(
         "interactive_agent_layer.config.config.get",
         lambda key, default=None: default,
     )
-    assert layer_config.get_scope_radius() == 3
+    assert layer_config.get_permission_timeout() == 120
 
 
-def test_get_scope_radius_from_config(monkeypatch):
-    """Reads agent_layer.scope_radius from config when set."""
+def test_get_permission_timeout_from_config(monkeypatch):
+    """Reads agent_layer.permission_timeout_seconds from config when set."""
     def fake_get(key, default=None):
-        if key == "agent_layer.scope_radius":
-            return 5
+        if key == "agent_layer.permission_timeout_seconds":
+            return 45
         return default
 
     monkeypatch.setattr("interactive_agent_layer.config.config.get", fake_get)
-    assert layer_config.get_scope_radius() == 5
+    assert layer_config.get_permission_timeout() == 45
+
+
+def test_get_scope_radius_removed():
+    """get_scope_radius() is deleted — replaced by envelope.load_permission_envelope()."""
+    assert not hasattr(layer_config, "get_scope_radius")
diff --git a/tests/interactive_agent_layer/test_envelope.py b/tests/interactive_agent_layer/test_envelope.py
new file mode 100644
index 00000000..b2b51ff1
--- /dev/null
+++ b/tests/interactive_agent_layer/test_envelope.py
@@ -0,0 +1,237 @@
+"""Tests for interactive_agent_layer.envelope — ScopeEnvelope + loaders + ⊆ validator."""
+from __future__ import annotations
+
+import pathlib
+
+import pytest
+import yaml
+
+from interactive_agent_layer.envelope import (
+    MAX_HOP_DISTANCE_BOUND,
+    ScopeEnvelope,
+    ScopeConfigError,
+    load_injection_envelope,
+    load_permission_envelope,
+    validate_injection_subset,
+)
+
+
+def _load_committed_app_config() -> dict:
+    """Load the committed config/app_config.yaml (not user overrides)."""
+    base = pathlib.Path(__file__).resolve()
+    for parent in base.parents:
+        candidate = parent / "config" / "app_config.yaml"
+        if candidate.exists():
+            with open(candidate) as f:
+                return yaml.safe_load(f)
+    raise FileNotFoundError("config/app_config.yaml not found from test location")
+
+
+# ---------------------------------------------------------------------------
+# ScopeEnvelope.m_allowed
+# ---------------------------------------------------------------------------
+
+def test_m_allowed_beyond_radius_is_none():
+    env = ScopeEnvelope(radius=3, m_max=4, decay=1)
+    assert env.m_allowed(4) is None
+
+
+def test_m_allowed_at_radius_boundary():
+    env = ScopeEnvelope(radius=3, m_max=4, decay=1)
+    assert env.m_allowed(3) == 1
+
+
+def test_m_allowed_at_zero_distance_is_m_max():
+    env = ScopeEnvelope(radius=3, m_max=4, decay=1)
+    assert env.m_allowed(0) == 4
+
+
+def test_m_allowed_clamped_to_one_minimum():
+    """decay*d can exceed m_max — result must never drop below 1 within radius."""
+    env = ScopeEnvelope(radius=5, m_max=4, decay=3)
+    assert env.m_allowed(2) == 1  # 4 - 3*2 = -2, clamped to 1
+
+
+def test_m_allowed_flat_curve_when_decay_zero():
+    env = ScopeEnvelope(radius=1, m_max=2, decay=0)
+    assert env.m_allowed(0) == 2
+    assert env.m_allowed(1) == 2
+
+
+def test_envelope_is_frozen():
+    env = ScopeEnvelope(radius=3, m_max=4, decay=1)
+    with pytest.raises(Exception):
+        env.radius = 10  # type: ignore[misc]
+
+
+# ---------------------------------------------------------------------------
+# validate_injection_subset — the ⊆ rule, including mid-range crossing
+# ---------------------------------------------------------------------------
+
+def test_validate_injection_subset_passes_for_current_defaults():
+    perm = ScopeEnvelope(radius=3, m_max=4, decay=1)
+    inj = ScopeEnvelope(radius=1, m_max=2, decay=0)
+    # Must not raise.
+    validate_injection_subset(inj, perm)
+
+
+def test_validate_injection_subset_radius_violation():
+    perm = ScopeEnvelope(radius=2, m_max=4, decay=1)
+    inj = ScopeEnvelope(radius=3, m_max=2, decay=0)
+    with pytest.raises(ScopeConfigError):
+        validate_injection_subset(inj, perm)
+
+
+def test_validate_injection_subset_endpoint_only_check_is_insufficient():
+    """Regression guard: curves that agree at d=0 and at inj.radius can still
+    cross in between. The validator must walk every d in [0, inj.radius], not
+    just the endpoints.
+
+    perm: radius=5, m_max=4, decay=3 -> [4, 1, 1, 1, 1, 1]
+    inj:  radius=4, m_max=3, decay=1 -> [3, 2, 1, 1, 1]
+
+    Endpoints (d=0: 3<=4, d=4: 1<=1) both pass, but d=1 (inj=2 > perm=1) fails.
+    """
+    perm = ScopeEnvelope(radius=5, m_max=4, decay=3)
+    inj = ScopeEnvelope(radius=4, m_max=3, decay=1)
+    with pytest.raises(ScopeConfigError):
+        validate_injection_subset(inj, perm)
+
+
+def test_validate_injection_subset_error_names_both_keys():
+    perm = ScopeEnvelope(radius=5, m_max=4, decay=3)
+    inj = ScopeEnvelope(radius=4, m_max=3, decay=1)
+    with pytest.raises(ScopeConfigError, match="permission"):
+        validate_injection_subset(inj, perm)
+
+
+# ---------------------------------------------------------------------------
+# load_permission_envelope / load_injection_envelope
+# ---------------------------------------------------------------------------
+
+def test_load_permission_envelope_defaults(monkeypatch):
+    monkeypatch.setattr(
+        "interactive_agent_layer.envelope.config.get",
+        lambda key, default=None: default,
+    )
+    env = load_permission_envelope()
+    assert env == ScopeEnvelope(radius=3, m_max=4, decay=1)
+
+
+def test_load_permission_envelope_from_config(monkeypatch):
+    def fake_get(key, default=None):
+        if key == "scope.permission":
+            return {"radius": 5, "m_max": 4, "decay": 1}
+        return default
+
+    monkeypatch.setattr("interactive_agent_layer.envelope.config.get", fake_get)
+    env = load_permission_envelope()
+    assert env == ScopeEnvelope(radius=5, m_max=4, decay=1)
+
+
+def test_load_permission_envelope_rejects_radius_beyond_hop_distance_bound(monkeypatch):
+    """A nonsensical radius > MAX_HOP_DISTANCE_BOUND fails fast at load time
+    instead of silently behaving as if capped (the DB query bounds max_N to
+    this same constant, so a larger configured radius would never actually
+    be honored — better to raise than silently under-enforce)."""
+    def fake_get(key, default=None):
+        if key == "scope.permission":
+            return {"radius": MAX_HOP_DISTANCE_BOUND + 1, "m_max": 4, "decay": 1}
+        return default
+
+    monkeypatch.setattr("interactive_agent_layer.envelope.config.get", fake_get)
+    with pytest.raises(ScopeConfigError):
+        load_permission_envelope()
+
+
+def test_load_permission_envelope_radius_at_bound_is_allowed(monkeypatch):
+    def fake_get(key, default=None):
+        if key == "scope.permission":
+            return {"radius": MAX_HOP_DISTANCE_BOUND, "m_max": 4, "decay": 1}
+        return default
+
+    monkeypatch.setattr("interactive_agent_layer.envelope.config.get", fake_get)
+    env = load_permission_envelope()
+    assert env.radius == MAX_HOP_DISTANCE_BOUND
+
+
+def test_load_permission_envelope_scenario_partial_merge(monkeypatch):
+    def fake_get(key, default=None):
+        if key == "scope.permission":
+            return {"radius": 3, "m_max": 4, "decay": 1}
+        if key == "scope.scenarios.beacon_autonomous.permission":
+            return {"radius": 5, "decay": 1}
+        return default
+
+    monkeypatch.setattr("interactive_agent_layer.envelope.config.get", fake_get)
+    env = load_permission_envelope(scenario="beacon_autonomous")
+    # radius/decay overridden, m_max inherited from the base envelope.
+    assert env == ScopeEnvelope(radius=5, m_max=4, decay=1)
+
+
+def test_load_injection_envelope_defaults(monkeypatch):
+    monkeypatch.setattr(
+        "interactive_agent_layer.envelope.config.get",
+        lambda key, default=None: default,
+    )
+    env = load_injection_envelope()
+    assert env == ScopeEnvelope(radius=1, m_max=2, decay=0)
+
+
+def test_load_injection_envelope_raises_on_subset_violation(monkeypatch):
+    def fake_get(key, default=None):
+        if key == "scope.permission":
+            return {"radius": 5, "m_max": 4, "decay": 3}
+        if key == "scope.injection":
+            return {"radius": 4, "m_max": 3, "decay": 1}
+        return default
+
+    monkeypatch.setattr("interactive_agent_layer.envelope.config.get", fake_get)
+    with pytest.raises(ScopeConfigError):
+        load_injection_envelope()
+
+
+def test_load_injection_envelope_validates_against_matching_scenario(monkeypatch):
+    """A scenario override on the permission side must be honored when
+    validating the injection envelope for that same scenario."""
+    def fake_get(key, default=None):
+        if key == "scope.permission":
+            return {"radius": 5, "m_max": 4, "decay": 1}
+        if key == "scope.injection":
+            return {"radius": 1, "m_max": 2, "decay": 0}
+        if key == "scope.scenarios.tight.permission":
+            return {"radius": 0}
+        if key == "scope.scenarios.tight.injection":
+            return {"radius": 0}
+        return default
+
+    monkeypatch.setattr("interactive_agent_layer.envelope.config.get", fake_get)
+    # Base defaults are fine (inj radius 1 <= perm radius 5).
+    load_injection_envelope()
+    # Scenario "tight" clamps injection radius to 0 too, still within its own
+    # scenario permission radius (0) -> should not raise.
+    load_injection_envelope(scenario="tight")
+
+
+# ---------------------------------------------------------------------------
+# The committed config/app_config.yaml scope: block
+# ---------------------------------------------------------------------------
+
+def test_committed_yaml_scope_block_matches_inert_defaults():
+    cfg = _load_committed_app_config()
+    scope = cfg["scope"]
+    assert scope["permission"] == {"radius": 3, "m_max": 4, "decay": 1}
+    assert scope["injection"] == {"radius": 1, "m_max": 2, "decay": 0}
+    assert scope["scenarios"] == {}
+
+
+def test_committed_yaml_scope_block_passes_subset_validation():
+    cfg = _load_committed_app_config()
+    perm = ScopeEnvelope(**cfg["scope"]["permission"])
+    inj = ScopeEnvelope(**cfg["scope"]["injection"])
+    validate_injection_subset(inj, perm)  # must not raise
+
+
+def test_committed_yaml_permission_timeout_is_120():
+    cfg = _load_committed_app_config()
+    assert cfg["agent_layer"]["permission_timeout_seconds"] == 120
diff --git a/tests/interactive_agent_layer/test_permissions.py b/tests/interactive_agent_layer/test_permissions.py
index 2eeff89c..bee1e447 100644
--- a/tests/interactive_agent_layer/test_permissions.py
+++ b/tests/interactive_agent_layer/test_permissions.py
@@ -263,6 +263,103 @@ async def test_permission_pending_cleaned_up_after_timeout(table, session, monke
 # 10. PermissionTimeoutError carries correct request_id
 # ---------------------------------------------------------------------------
 
+# ---------------------------------------------------------------------------
+# 11. permission_resolved — emitted on every terminal resolution
+# ---------------------------------------------------------------------------
+
+async def _drain_all(queue: asyncio.Queue, task: asyncio.Task) -> list[dict]:
+    """Collect every event put on queue until task completes."""
+    events: list[dict] = []
+    while not task.done():
+        get_task = asyncio.ensure_future(queue.get())
+        done, _ = await asyncio.wait([get_task, task], return_when=asyncio.FIRST_COMPLETED)
+        if get_task in done:
+            events.append(get_task.result())
+        else:
+            get_task.cancel()
+    # Drain anything left in the queue after task completion.
+    while not queue.empty():
+        events.append(queue.get_nowait())
+    return events
+
+
+async def test_permission_resolved_emitted_on_approval(table, session):
+    queue: asyncio.Queue = asyncio.Queue()
+    gate = PermissionGate(
+        translation_table=table, session=session, outbound_events=queue,
+        auto_approve_user_actions=False,
+    )
+    task = asyncio.create_task(gate.can_use_tool("upsert_tasks", {"count": 1, "tasks": []}, None))
+    request_event = await asyncio.wait_for(queue.get(), timeout=5.0)
+    fut = session.permission_pending.get(request_event["request_id"])
+    fut.set_result(True)
+    await task
+    resolved = await asyncio.wait_for(queue.get(), timeout=5.0)
+    assert resolved["type"] == "permission_resolved"
+    assert resolved["request_id"] == request_event["request_id"]
+    assert resolved["resolution"] == "approved"
+
+
+async def test_permission_resolved_emitted_on_denial(table, session):
+    queue: asyncio.Queue = asyncio.Queue()
+    gate = PermissionGate(
+        translation_table=table, session=session, outbound_events=queue,
+        auto_approve_user_actions=False,
+    )
+    task = asyncio.create_task(gate.can_use_tool("upsert_tasks", {"count": 1, "tasks": []}, None))
+    request_event = await asyncio.wait_for(queue.get(), timeout=5.0)
+    fut = session.permission_pending.get(request_event["request_id"])
+    fut.set_result(False)
+    await task
+    resolved = await asyncio.wait_for(queue.get(), timeout=5.0)
+    assert resolved["type"] == "permission_resolved"
+    assert resolved["resolution"] == "denied"
+
+
+async def test_permission_resolved_emitted_on_timeout_before_raise(table, session, monkeypatch):
+    """Write-kind timeout still emits permission_resolved before raising."""
+    monkeypatch.setattr(
+        "interactive_agent_layer.permissions.get_permission_timeout",
+        lambda: 0.01,
+    )
+    queue: asyncio.Queue = asyncio.Queue()
+    gate = PermissionGate(
+        translation_table=table, session=session, outbound_events=queue,
+        auto_approve_user_actions=False,
+    )
+    with pytest.raises(PermissionTimeoutError):
+        await gate.can_use_tool("upsert_tasks", {"count": 1, "tasks": []}, None)
+    events = []
+    while not queue.empty():
+        events.append(queue.get_nowait())
+    kinds = [e["type"] for e in events]
+    assert "permission_request" in kinds
+    resolved = [e for e in events if e["type"] == "permission_resolved"]
+    assert len(resolved) == 1
+    assert resolved[0]["resolution"] == "timeout"
+
+
+# ---------------------------------------------------------------------------
+# 12. reason_from_bot plumbed from optional tool-call 'reason' arg
+# ---------------------------------------------------------------------------
+
+async def test_reason_from_bot_plumbed_from_tool_args(table, session):
+    result, event = await _run_with_queue(
+        table, session, "upsert_tasks",
+        {"count": 1, "tasks": [], "reason": "user asked for it"},
+        resolve_value=True,
+    )
+    assert event["reason_from_bot"] == "user asked for it"
+
+
+async def test_reason_from_bot_defaults_to_none(table, session):
+    result, event = await _run_with_queue(
+        table, session, "upsert_tasks", {"count": 1, "tasks": []},
+        resolve_value=True,
+    )
+    assert event["reason_from_bot"] is None
+
+
 async def test_permission_timeout_error_request_id_matches_event(table, session, monkeypatch):
     """PermissionTimeoutError.request_id matches the request_id emitted in the event."""
     monkeypatch.setattr(
diff --git a/tests/interactive_agent_layer/test_scope_gating.py b/tests/interactive_agent_layer/test_scope_gating.py
index 8974cc14..0e4acdef 100644
--- a/tests/interactive_agent_layer/test_scope_gating.py
+++ b/tests/interactive_agent_layer/test_scope_gating.py
@@ -1,4 +1,9 @@
-"""Tests for read_context scope gating via PermissionGate (TDD — written before implementation)."""
+"""Tests for read_context scope gating via PermissionGate + ScopeEnvelope.
+
+Uses ScopeEnvelope(radius=2, m_max=4, decay=0) for most cases (flat M=4
+within radius, matching the old flat scope_radius=2 behavior) plus dedicated
+graded-M cases that exercise the decay curve.
+"""
 from __future__ import annotations
 
 import asyncio
@@ -6,6 +11,7 @@
 
 import pytest
 
+from interactive_agent_layer.envelope import ScopeEnvelope
 from interactive_agent_layer.permissions import (
     PermissionGate,
     PermissionResultAllow,
@@ -14,6 +20,8 @@
 from interactive_agent_layer.session import Session
 from interactive_agent_layer.translation import TranslationTable
 
+FLAT_ENVELOPE = ScopeEnvelope(radius=2, m_max=4, decay=0)
+
 
 @pytest.fixture(autouse=True)
 def patch_permission_timeout(monkeypatch):
@@ -42,7 +50,7 @@ def session():
         agent_version="tether-agent-2.0",
         options={
             "scope_source_node_id": "node-source",
-            "scope_radius": 2,
+            "permission_envelope": FLAT_ENVELOPE,
         },
     )
 
@@ -51,10 +59,12 @@ def _make_gate(
     table,
     session,
     hop_fn=None,
-    scope_mode="tree_distance",
+    envelope=FLAT_ENVELOPE,
     resolve_path_fn=None,
     *,
     auto_approve=False,
+    check_grant_fn=None,
+    insert_grant_fn=None,
 ):
     return PermissionGate(
         translation_table=table,
@@ -62,17 +72,22 @@ def _make_gate(
         outbound_events=asyncio.Queue(),
         auto_approve_user_actions=auto_approve,
         scope_source_node_id=session.options.get("scope_source_node_id"),
-        scope_radius=session.options.get("scope_radius"),
-        scope_mode=scope_mode,
+        scope_envelope=envelope,
         hop_distance_fn=hop_fn,
         resolve_node_path_fn=resolve_path_fn,
+        check_grant_fn=check_grant_fn,
+        insert_grant_fn=insert_grant_fn,
     )
 
 
-async def _run_scope_check(table, session, args, hop_fn, resolve_path_fn=None):
+async def _run_scope_check(
+    table, session, args, hop_fn, resolve_path_fn=None,
+    envelope=FLAT_ENVELOPE, check_grant_fn=None, insert_grant_fn=None,
+):
     """Run can_use_tool for read_context, intercept the permission_request event.
 
-    Returns (task, event, session) — caller must resolve the future before awaiting task.
+    Returns (task, event, queue) — caller must resolve the future before
+    awaiting task, and may drain `queue` afterward for permission_resolved.
     """
     queue: asyncio.Queue = asyncio.Queue()
     gate = PermissionGate(
@@ -81,13 +96,15 @@ async def _run_scope_check(table, session, args, hop_fn, resolve_path_fn=None):
         outbound_events=queue,
         auto_approve_user_actions=False,
         scope_source_node_id=session.options.get("scope_source_node_id"),
-        scope_radius=session.options.get("scope_radius"),
+        scope_envelope=envelope,
         hop_distance_fn=hop_fn,
         resolve_node_path_fn=resolve_path_fn,
+        check_grant_fn=check_grant_fn,
+        insert_grant_fn=insert_grant_fn,
     )
     task = asyncio.create_task(gate.can_use_tool("read_context", args, None))
     event = await asyncio.wait_for(queue.get(), timeout=5.0)
-    return task, event
+    return task, event, queue
 
 
 # ---------------------------------------------------------------------------
@@ -109,14 +126,14 @@ async def test_read_context_no_scope_source_always_allow(table, session):
 
 
 async def test_read_context_hop_fn_none_always_allow(table, session):
-    """When hop_distance_fn is None (even if source is set), gate is inactive."""
+    """When hop_distance_fn is None (even if source/envelope are set), gate is inactive."""
     gate = PermissionGate(
         translation_table=table,
         session=session,
         outbound_events=asyncio.Queue(),
         auto_approve_user_actions=False,
         scope_source_node_id="node-source",
-        scope_radius=2,
+        scope_envelope=FLAT_ENVELOPE,
         hop_distance_fn=None,
     )
     result = await gate.can_use_tool("read_context", {"node_ids": ["node-far-away"]}, None)
@@ -140,7 +157,7 @@ async def hop_fn(from_id, to_id):
 
 
 async def test_read_context_in_scope_node_id_allows(table, session):
-    """Distance ≤ M → allow without prompting."""
+    """Distance ≤ radius → allow without prompting."""
 
     async def hop_fn(from_id, to_id):
         return 1  # within radius 2
@@ -165,7 +182,7 @@ async def test_read_context_all_targets_in_scope_allows(table, session):
     """All node_ids within scope → allow without prompting."""
 
     async def hop_fn(from_id, to_id):
-        return 2  # exactly at M boundary — in scope
+        return 2  # exactly at radius boundary — in scope
 
     gate = _make_gate(table, session, hop_fn)
     result = await gate.can_use_tool(
@@ -180,12 +197,12 @@ async def hop_fn(from_id, to_id):
 
 
 async def test_read_context_out_of_scope_emits_permission_request(table, session):
-    """Distance > M → permission_request with kind='read_out_of_scope' emitted."""
+    """Distance > radius → permission_request with kind='read_out_of_scope' emitted."""
 
     async def hop_fn(from_id, to_id):
         return 5  # exceeds radius 2
 
-    task, event = await _run_scope_check(
+    task, event, _ = await _run_scope_check(
         table, session, {"node_ids": ["node-distant"]}, hop_fn
     )
     request_id = event["request_id"]
@@ -206,7 +223,7 @@ async def test_read_context_out_of_scope_user_denies(table, session):
     async def hop_fn(from_id, to_id):
         return 5
 
-    task, event = await _run_scope_check(
+    task, event, _ = await _run_scope_check(
         table, session, {"node_ids": ["node-distant"]}, hop_fn
     )
     request_id = event["request_id"]
@@ -224,7 +241,7 @@ async def test_read_context_none_distance_treated_as_out_of_scope(table, session
     async def hop_fn(from_id, to_id):
         return None  # unrelated trees
 
-    task, event = await _run_scope_check(
+    task, event, _ = await _run_scope_check(
         table, session, {"node_ids": ["node-unrelated"]}, hop_fn
     )
     request_id = event["request_id"]
@@ -243,7 +260,7 @@ async def test_read_context_first_offender_is_reported_in_target(table, session)
     async def hop_fn(from_id, to_id):
         return 1 if to_id == "node-close" else 5
 
-    task, event = await _run_scope_check(
+    task, event, _ = await _run_scope_check(
         table, session,
         {"node_ids": ["node-close", "node-distant"]},
         hop_fn,
@@ -272,7 +289,7 @@ async def hop_fn(from_id, to_id):
     async def resolve_path(path):
         return "node-resolved-from-path"
 
-    task, event = await _run_scope_check(
+    task, event, _ = await _run_scope_check(
         table, session,
         {"paths": ["Projects/OutOfScope"]},
         hop_fn,
@@ -296,7 +313,7 @@ async def hop_fn(from_id, to_id):
     async def resolve_path(path):
         return None  # path not found
 
-    task, event = await _run_scope_check(
+    task, event, _ = await _run_scope_check(
         table, session,
         {"paths": ["NonExistent/Ghost"]},
         hop_fn,
@@ -327,6 +344,198 @@ async def resolve_path(path):
     assert isinstance(result, PermissionResultAllow)
 
 
+# ---------------------------------------------------------------------------
+# D: Graded envelope — requested_M vs m_allowed(d)
+# ---------------------------------------------------------------------------
+
+GRADED_ENVELOPE = ScopeEnvelope(radius=3, m_max=4, decay=1)  # m_allowed: 4,3,2,1
+
+
+async def test_read_context_in_radius_but_M_exceeds_m_allowed_gates(table, session):
+    """d=2 (within radius 3) but requested M=4 > m_allowed(2)=2 → gated."""
+
+    async def hop_fn(from_id, to_id):
+        return 2
+
+    task, event, _ = await _run_scope_check(
+        table, session, {"node_ids": ["node-mid"], "M": 4}, hop_fn,
+        envelope=GRADED_ENVELOPE,
+    )
+    request_id = event["request_id"]
+    fut = session.permission_pending.get(request_id)
+    fut.set_result(True)
+    await task
+
+    assert event.get("kind") == "read_out_of_scope"
+
+
+async def test_read_context_in_radius_M_within_m_allowed_allows(table, session):
+    """d=2, requested M=2 <= m_allowed(2)=2 → allowed without prompting."""
+
+    async def hop_fn(from_id, to_id):
+        return 2
+
+    gate = _make_gate(table, session, hop_fn, envelope=GRADED_ENVELOPE)
+    result = await gate.can_use_tool(
+        "read_context", {"node_ids": ["node-mid"], "M": 2}, None
+    )
+    assert isinstance(result, PermissionResultAllow)
+
+
+async def test_read_context_default_M_is_4(table, session):
+    """Omitting M defaults to 4 (matches read_context's own tool default)."""
+
+    async def hop_fn(from_id, to_id):
+        return 0  # m_allowed(0) = 4, so default M=4 must pass
+
+    gate = _make_gate(table, session, hop_fn, envelope=GRADED_ENVELOPE)
+    result = await gate.can_use_tool("read_context", {"node_ids": ["node-source"]}, None)
+    assert isinstance(result, PermissionResultAllow)
+
+
+async def test_read_context_beyond_radius_still_gates_regardless_of_M(table, session):
+    """d=4 > radius 3 → gated even at the lowest M=1."""
+
+    async def hop_fn(from_id, to_id):
+        return 4
+
+    task, event, _ = await _run_scope_check(
+        table, session, {"node_ids": ["node-far"], "M": 1}, hop_fn,
+        envelope=GRADED_ENVELOPE,
+    )
+    fut = session.permission_pending.get(event["request_id"])
+    fut.set_result(True)
+    await task
+    assert event.get("kind") == "read_out_of_scope"
+
+
+# ---------------------------------------------------------------------------
+# E: Grant check/insert on the scope-read path (DD 3.2 gap fix)
+# ---------------------------------------------------------------------------
+
+
+async def test_scope_grant_exists_skips_permission_request(table, session):
+    """check_grant_fn returning True auto-allows without emitting permission_request."""
+
+    async def hop_fn(from_id, to_id):
+        return 5  # out of scope
+
+    async def has_grant(user_id, conversation_id, target, kind):
+        assert kind == "read_out_of_scope"
+        return True
+
+    gate = _make_gate(table, session, hop_fn, check_grant_fn=has_grant)
+    result = await gate.can_use_tool("read_context", {"node_ids": ["node-distant"]}, None)
+    assert isinstance(result, PermissionResultAllow)
+
+
+async def test_scope_grant_inserted_on_approval(table, session):
+    """On user approval, insert_grant_fn is called with (user_id, conv_id, target, kind)."""
+
+    async def hop_fn(from_id, to_id):
+        return 5
+
+    async def no_grant(user_id, conversation_id, target, kind):
+        return False
+
+    inserted: list[dict] = []
+
+    async def insert_grant(user_id, conversation_id, target, kind):
+        inserted.append({"user_id": user_id, "conv": conversation_id,
+                         "target": target, "kind": kind})
+
+    task, event, _ = await _run_scope_check(
+        table, session, {"node_ids": ["node-distant"]}, hop_fn,
+        check_grant_fn=no_grant, insert_grant_fn=insert_grant,
+    )
+    fut = session.permission_pending.get(event["request_id"])
+    fut.set_result(True)
+    await task
+
+    assert len(inserted) == 1
+    assert inserted[0]["kind"] == "read_out_of_scope"
+    assert inserted[0]["target"] == "node-distant"
+
+
+async def test_scope_grant_not_inserted_on_denial(table, session):
+    """On user denial, insert_grant_fn is NOT called."""
+
+    async def hop_fn(from_id, to_id):
+        return 5
+
+    async def no_grant(user_id, conversation_id, target, kind):
+        return False
+
+    inserted: list = []
+
+    async def insert_grant(user_id, conversation_id, target, kind):
+        inserted.append(kind)
+
+    task, event, _ = await _run_scope_check(
+        table, session, {"node_ids": ["node-distant"]}, hop_fn,
+        check_grant_fn=no_grant, insert_grant_fn=insert_grant,
+    )
+    fut = session.permission_pending.get(event["request_id"])
+    fut.set_result(False)
+    await task
+
+    assert inserted == []
+
+
+# ---------------------------------------------------------------------------
+# F: permission_resolved on the scope-read path
+# ---------------------------------------------------------------------------
+
+
+async def test_scope_permission_resolved_emitted_on_approval(table, session):
+    async def hop_fn(from_id, to_id):
+        return 5
+
+    task, event, queue = await _run_scope_check(
+        table, session, {"node_ids": ["node-distant"]}, hop_fn
+    )
+    fut = session.permission_pending.get(event["request_id"])
+    fut.set_result(True)
+    await task
+
+    resolved = await asyncio.wait_for(queue.get(), timeout=5.0)
+    assert resolved["type"] == "permission_resolved"
+    assert resolved["request_id"] == event["request_id"]
+    assert resolved["resolution"] == "approved"
+
+
+async def test_scope_permission_resolved_emitted_on_timeout(table, session, monkeypatch):
+    """Read-kind timeout resolves as deny AND still emits permission_resolved
+    (no exception raised — reads deny-and-continue per DD §4.5)."""
+    monkeypatch.setattr(
+        "interactive_agent_layer.permissions.get_permission_timeout",
+        lambda: 0.01,
+    )
+
+    async def hop_fn(from_id, to_id):
+        return 5
+
+    queue: asyncio.Queue = asyncio.Queue()
+    gate = PermissionGate(
+        translation_table=table,
+        session=session,
+        outbound_events=queue,
+        auto_approve_user_actions=False,
+        scope_source_node_id="node-source",
+        scope_envelope=FLAT_ENVELOPE,
+        hop_distance_fn=hop_fn,
+    )
+    result = await gate.can_use_tool("read_context", {"node_ids": ["node-far"]}, None)
+    assert isinstance(result, PermissionResultDeny)
+
+    events = []
+    while not queue.empty():
+        events.append(queue.get_nowait())
+    resolved = [e for e in events if e["type"] == "permission_resolved"]
+    assert len(resolved) == 1
+    assert resolved[0]["resolution"] == "timeout"
+
+
 # ---------------------------------------------------------------------------
 # Scope check does not affect other tools
 # ---------------------------------------------------------------------------
@@ -355,7 +564,7 @@ async def test_scope_gate_pending_cleaned_up_after_approve(table, session):
     async def hop_fn(from_id, to_id):
         return 5
 
-    task, event = await _run_scope_check(
+    task, event, _ = await _run_scope_check(
         table, session, {"node_ids": ["node-far"]}, hop_fn
     )
     request_id = event["request_id"]
@@ -384,7 +593,7 @@ async def hop_fn(from_id, to_id):
         outbound_events=queue,
         auto_approve_user_actions=False,
         scope_source_node_id="node-source",
-        scope_radius=2,
+        scope_envelope=FLAT_ENVELOPE,
         hop_distance_fn=hop_fn,
     )
     result = await gate.can_use_tool("read_context", {"node_ids": ["node-far"]}, None)
diff --git a/tests/interactive_agent_layer/test_session.py b/tests/interactive_agent_layer/test_session.py
index 399a3f5f..087edeb0 100644
--- a/tests/interactive_agent_layer/test_session.py
+++ b/tests/interactive_agent_layer/test_session.py
@@ -5,6 +5,7 @@
 
 import pytest
 
+from interactive_agent_layer.envelope import ScopeEnvelope
 from interactive_agent_layer.session import Layer, Session
 from interactive_agent_layer.ws_publisher import WSPublisher
 
@@ -478,7 +479,7 @@ def _scope_requests(events: list[dict]) -> list[dict]:
 
 
 async def test_run_turn_wires_scope_from_session_options(monkeypatch):
-    """scope_source_node_id/scope_radius in session.options + Layer.hop_distance_fn
+    """scope_source_node_id/permission_envelope in session.options + Layer.hop_distance_fn
     activate scope gating: an out-of-range read_context target emits
     read_out_of_scope and, on timeout, is denied."""
     monkeypatch.setattr(
@@ -497,7 +498,10 @@ async def hop_distance_fn(user_id, from_id, to_id):
 
     s = layer.create_session(
         "user1", "wsid1", "v1",
-        options={"scope_source_node_id": "source-node", "scope_radius": 2},
+        options={
+            "scope_source_node_id": "source-node",
+            "permission_envelope": {"radius": 2, "m_max": 4, "decay": 0},
+        },
     )
 
     events = []
@@ -521,7 +525,10 @@ async def hop_distance_fn(user_id, from_id, to_id):
 
     s = layer.create_session(
         "user1", "wsid1", "v1",
-        options={"scope_source_node_id": "source-node", "scope_radius": 2},
+        options={
+            "scope_source_node_id": "source-node",
+            "permission_envelope": {"radius": 2, "m_max": 4, "decay": 0},
+        },
     )
 
     events = []
@@ -549,12 +556,13 @@ async def test_run_turn_no_scope_config_is_backwards_compatible():
 
 async def test_run_turn_resolves_scope_from_conversation_id(monkeypatch):
     """No explicit scope options — resolve_conversation_scope_fn resolves
-    scope_source_node_id from conversation_id, scope_radius from config."""
+    scope_source_node_id from conversation_id, scope_envelope from config."""
     monkeypatch.setattr(
         "interactive_agent_layer.permissions.get_permission_timeout", lambda: 0.01
     )
     monkeypatch.setattr(
-        "interactive_agent_layer.session.get_scope_radius", lambda: 2
+        "interactive_agent_layer.session.load_permission_envelope",
+        lambda: ScopeEnvelope(radius=2, m_max=4, decay=0),
     )
 
     pool = _ReadContextControlPool(node_ids=["far-node"])
@@ -585,6 +593,45 @@ async def hop_distance_fn(user_id, from_id, to_id):
     assert len(_scope_requests(events)) == 1
 
 
+async def test_run_turn_permission_envelope_option_overrides_config_default(monkeypatch):
+    """session.options['permission_envelope'] wins over load_permission_envelope()
+    — whole-envelope override, caller-supplied at spawn (never model-supplied)."""
+    monkeypatch.setattr(
+        "interactive_agent_layer.permissions.get_permission_timeout", lambda: 0.01
+    )
+
+    def _unexpected_config_default():
+        raise AssertionError("load_permission_envelope() must not be called when option is set")
+
+    monkeypatch.setattr(
+        "interactive_agent_layer.session.load_permission_envelope",
+        _unexpected_config_default,
+    )
+
+    pool = _ReadContextControlPool(node_ids=["near-node"])
+
+    async def hop_distance_fn(user_id, from_id, to_id):
+        return 1  # within the override's radius of 5
+
+    layer = _make_layer(pool)
+    layer.hop_distance_fn = hop_distance_fn
+
+    s = layer.create_session(
+        "user1", "wsid1", "v1",
+        options={
+            "scope_source_node_id": "source-node",
+            "permission_envelope": {"radius": 5, "m_max": 4, "decay": 0},
+        },
+    )
+
+    events = []
+    async for event in layer.run_turn(s.session_id, "hi"):
+        events.append(event)
+
+    assert _scope_requests(events) == []
+    assert pool.decisions == ["allow"]
+
+
 async def test_run_turn_conversation_scope_resolves_to_none_disables_gating():
     """Conversation has no context_node_id → resolve returns None → no gating."""
     pool = _ReadContextControlPool(node_ids=["any-node"])
diff --git a/tests/mcp/test_read_context_enforcement.py b/tests/mcp/test_read_context_enforcement.py
index b71b3f2b..5913022d 100644
--- a/tests/mcp/test_read_context_enforcement.py
+++ b/tests/mcp/test_read_context_enforcement.py
@@ -65,9 +65,11 @@ def _patch_all(mocks: dict):
         # get_node_tree_distance must not exist as a call path anymore; if the
         # implementation still imports/calls it, patching it as a stub that
         # raises makes any accidental reintroduction fail loudly.
+        # (Lives in db.pg_queries.nodes as of the distance-query consolidation
+        # — moved from node_memory, generalized to multi-root.)
         stack.enter_context(
             patch(
-                "db.pg_queries.node_memory.get_node_tree_distance",
+                "db.pg_queries.nodes.get_node_tree_distance",
                 AsyncMock(side_effect=AssertionError(
                     "execute_read_context must not call get_node_tree_distance "
                     "(scope enforcement lives solely in PermissionGate)"