Brief 1b: PermissionGate sole enforcer + graded 2D scope envelope

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: {}

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,
 )

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,

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 ─────────────────────────────────────────────────────────

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))

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)"
+            )