From 0f49e51a34445404566b2cca247f2c491a268995 Mon Sep 17 00:00:00 2001
From: Jason Lunder <jlunder@zagmail.gonzaga.edu>
Date: Sat, 4 Jul 2026 11:28:53 -0700
Subject: [PATCH 1/3] wip: RLS hardening addendum for BRIEF 1a (node_read_log
 trio + get_node_by_path)

Adds optional user_id param to log_node_read, has_read_node_in_conversation,
get_conversation_reads (db/pg_queries/node_memory.py) and get_node_by_path
(db/pg_queries/nodes.py). When provided, binds user_id as an explicit
WHERE/VALUES filter instead of relying solely on the app.current_user_id
session GUC -- defense-in-depth, correct even on an unscoped connection.
When None (default), behavior unchanged (RLS-only), backward compatible.

REMAINING (not done in this commit):
- Bind user_id=get_user_id() at call sites in tether_mcp/tools/read_context.py,
  read_node_memory.py, write_node_memory.py, server.py.
- Main BRIEF 1a change: conversation_id optional in execute_read_context,
  delete out_of_scope error dicts + get_node_tree_distance calls in
  execute_read_context (~243-268) and _add_children (~163-173), rename
  N -> traverse_depth.
- server.py tool docstring rewrite (no authorization claims).
- Test updates: tests/mcp/test_read_context_enforcement.py needs a full
  rewrite (its whole premise -- conversation_id required, out_of_scope
  errors -- is being removed). tests/tether_mcp/test_read_context_cascade.py
  needs get_node_tree_distance mocks removed + N->traverse_depth rename +
  delete test_conversation_id_required.
- REQUIRED regression test: real (non-mocked) execute_read_context with
  conversation_id=None returning real root nodes -- not yet written.
- RLS hardening acceptance test (mismatched user_id returns nothing on
  unscoped connection) -- not yet written.

See cc-context-store/tether/docs/handoffs/neon-audit-premium-handoff-2026-07-04.md
for full status and matched-pair contract with the premium side.
---
 db/pg_queries/node_memory.py | 114 ++++++++++++++++++++++++++---------
 db/pg_queries/nodes.py       |  42 ++++++++++---
 2 files changed, 119 insertions(+), 37 deletions(-)

diff --git a/db/pg_queries/node_memory.py b/db/pg_queries/node_memory.py
index 161af882..cb78445b 100644
--- a/db/pg_queries/node_memory.py
+++ b/db/pg_queries/node_memory.py
@@ -110,46 +110,83 @@ async def log_node_read(
     *,
     conversation_id: str | None = None,
     title: str | None = None,
+    user_id: str | None = None,
 ) -> None:
     """Record that the current user read node_id at level_ordinal in this conversation.
 
     Called by read_context and read_node_memory on every access.
     conversation_id may be None for admin/debug calls without conversation context.
+
+    user_id: RLS hardening — explicit caller-supplied user_id, inserted directly
+    instead of relying solely on the `app.current_user_id` session GUC. When
+    None (default), falls back to the session GUC as before, so this is
+    backward compatible for any caller that hasn't been updated to bind it.
     """
     conv_uuid = _uuid.UUID(conversation_id) if conversation_id else None
-    await conn.execute(
-        """
-        INSERT INTO node_read_log (user_id, conversation_id, node_id, level_ordinal, title)
-        VALUES (
-            current_setting('app.current_user_id', true)::uuid,
-            $1, $2::uuid, $3, $4
+    if user_id is not None:
+        await conn.execute(
+            """
+            INSERT INTO node_read_log (user_id, conversation_id, node_id, level_ordinal, title)
+            VALUES ($1::uuid, $2, $3::uuid, $4, $5)
+            """,
+            _uuid.UUID(user_id), conv_uuid, _uuid.UUID(node_id), level_ordinal, title,
+        )
+    else:
+        await conn.execute(
+            """
+            INSERT INTO node_read_log (user_id, conversation_id, node_id, level_ordinal, title)
+            VALUES (
+                current_setting('app.current_user_id', true)::uuid,
+                $1, $2::uuid, $3, $4
+            )
+            """,
+            conv_uuid, _uuid.UUID(node_id), level_ordinal, title,
         )
-        """,
-        conv_uuid, _uuid.UUID(node_id), level_ordinal, title,
-    )
 
 
 async def has_read_node_in_conversation(
     conn: asyncpg.Connection,
     node_id: str,
     conversation_id: str,
+    *,
+    user_id: str | None = None,
 ) -> bool:
     """Return True if any read of node_id was logged for this conversation.
 
     Used by write_node_memory advisory read-before-write check:
       v1: log WARNING if False but allow the write
       v2: hard block if False once Stream C callers always provide conversation_id
+
+    user_id: RLS hardening — when provided, the query binds this value
+    directly as an explicit `WHERE user_id = $N` filter instead of relying
+    on the `app.current_user_id` session GUC. This makes the scoping correct
+    even on an "unscoped" connection (no `SET LOCAL app.current_user_id`
+    applied) where the GUC-based filter would either be wrong or raise a
+    cast error on an empty string. When None (default), behavior is
+    unchanged (session-GUC only) for backward compatibility.
     """
-    count = await conn.fetchval(
-        """
-        SELECT COUNT(*)
-        FROM node_read_log
-        WHERE user_id         = current_setting('app.current_user_id', true)::uuid
-          AND conversation_id = $1::uuid
-          AND node_id         = $2::uuid
-        """,
-        _uuid.UUID(conversation_id), _uuid.UUID(node_id),
-    )
+    if user_id is not None:
+        count = await conn.fetchval(
+            """
+            SELECT COUNT(*)
+            FROM node_read_log
+            WHERE user_id         = $1::uuid
+              AND conversation_id = $2::uuid
+              AND node_id         = $3::uuid
+            """,
+            _uuid.UUID(user_id), _uuid.UUID(conversation_id), _uuid.UUID(node_id),
+        )
+    else:
+        count = await conn.fetchval(
+            """
+            SELECT COUNT(*)
+            FROM node_read_log
+            WHERE user_id         = current_setting('app.current_user_id', true)::uuid
+              AND conversation_id = $1::uuid
+              AND node_id         = $2::uuid
+            """,
+            _uuid.UUID(conversation_id), _uuid.UUID(node_id),
+        )
     return (count or 0) > 0
 
 
@@ -207,19 +244,38 @@ async def get_node_tree_distance(
 async def get_conversation_reads(
     conn: asyncpg.Connection,
     conversation_id: str,
+    *,
+    user_id: str | None = None,
 ) -> list[dict]:
     """Return all node read credits for a conversation (newest first).
 
     Used for diagnostics and advisory enforcement reports.
+
+    user_id: RLS hardening — when provided, binds this value directly as an
+    explicit filter instead of relying on the session GUC (see
+    has_read_node_in_conversation for the full rationale). When None
+    (default), behavior is unchanged (session-GUC only).
     """
-    rows = await conn.fetch(
-        """
-        SELECT node_id::text, level_ordinal, title, read_at
-        FROM node_read_log
-        WHERE user_id         = current_setting('app.current_user_id', true)::uuid
-          AND conversation_id = $1::uuid
-        ORDER BY read_at DESC
-        """,
-        _uuid.UUID(conversation_id),
-    )
+    if user_id is not None:
+        rows = await conn.fetch(
+            """
+            SELECT node_id::text, level_ordinal, title, read_at
+            FROM node_read_log
+            WHERE user_id         = $1::uuid
+              AND conversation_id = $2::uuid
+            ORDER BY read_at DESC
+            """,
+            _uuid.UUID(user_id), _uuid.UUID(conversation_id),
+        )
+    else:
+        rows = await conn.fetch(
+            """
+            SELECT node_id::text, level_ordinal, title, read_at
+            FROM node_read_log
+            WHERE user_id         = current_setting('app.current_user_id', true)::uuid
+              AND conversation_id = $1::uuid
+            ORDER BY read_at DESC
+            """,
+            _uuid.UUID(conversation_id),
+        )
     return [dict(r) for r in rows]
diff --git a/db/pg_queries/nodes.py b/db/pg_queries/nodes.py
index 22e1593a..4a7a8680 100644
--- a/db/pg_queries/nodes.py
+++ b/db/pg_queries/nodes.py
@@ -65,21 +65,47 @@ async def get_node(conn: asyncpg.Connection, node_id: str) -> dict | None:
     return d
 
 
-async def get_node_by_path(conn: asyncpg.Connection, path: str) -> dict | None:
+async def get_node_by_path(
+    conn: asyncpg.Connection,
+    path: str,
+    *,
+    user_id: str | None = None,
+) -> dict | None:
+    """Resolve a slash-separated path (e.g. "Projects/Tether") to a node.
+
+    user_id: RLS hardening — when provided, each step's lookup binds this
+    value directly as an explicit `AND user_id = $N` filter instead of
+    relying solely on RLS/the session GUC. This makes path resolution
+    correct even on an unscoped connection. When None (default), behavior
+    is unchanged (RLS-only).
+    """
     parts = [p for p in path.split("/") if p]
     if not parts:
         return None
-    row = await conn.fetchrow(
-        "SELECT id FROM context_nodes WHERE parent_id IS NULL AND name = $1", parts[0]
-    )
+    uid = _uuid.UUID(user_id) if user_id is not None else None
+    if uid is not None:
+        row = await conn.fetchrow(
+            "SELECT id FROM context_nodes WHERE parent_id IS NULL AND name = $1 AND user_id = $2::uuid",
+            parts[0], uid,
+        )
+    else:
+        row = await conn.fetchrow(
+            "SELECT id FROM context_nodes WHERE parent_id IS NULL AND name = $1", parts[0]
+        )
     if not row:
         return None
     resolved_id = row["id"]
     for segment in parts[1:]:
-        row = await conn.fetchrow(
-            "SELECT id FROM context_nodes WHERE parent_id = $1 AND name = $2",
-            resolved_id, segment,
-        )
+        if uid is not None:
+            row = await conn.fetchrow(
+                "SELECT id FROM context_nodes WHERE parent_id = $1 AND name = $2 AND user_id = $3::uuid",
+                resolved_id, segment, uid,
+            )
+        else:
+            row = await conn.fetchrow(
+                "SELECT id FROM context_nodes WHERE parent_id = $1 AND name = $2",
+                resolved_id, segment,
+            )
         if not row:
             return None
         resolved_id = row["id"]

From e6dc6eb56e517d5fc657ec11700b4357b11f192e Mon Sep 17 00:00:00 2001
From: Jason Lunder <jlunder@zagmail.gonzaga.edu>
Date: Sat, 4 Jul 2026 18:31:48 -0700
Subject: [PATCH 2/3] demote read_context to pure retrieval; RLS user_id
 hardening at call sites
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

read_context is no longer a scope enforcer. PermissionGate
(interactive_agent_layer/permissions.py) is the sole enforcer per the
conversational-core design review (5.1/5.2); by the time a read_context
call reaches this module the gate has already judged whether it should
happen. This PR:

- Deletes the conversation_id_required error dict — conversation_id is
  now fully optional. Absent conversation_id means no read-credit
  logging, not a refusal.
- Deletes the out_of_scope refusal machinery (get_node_tree_distance
  calls + error dicts) from execute_read_context and _add_children.
  Cascade descent is still cost-bounded, just never produces an error.
- Renames N -> traverse_depth throughout (tool signature, internals,
  tests) to make clear it's a cost bound, not an authorization radius.
- server.py docstring rewritten to drop all authorization claims for
  read_context; M is a detail request the gate judges, not enforced
  here.
- write_node_memory's read-before-write enforcement is UNCHANGED (still
  requires conversation_id + a prior logged read).

RLS hardening (0e addendum, folded into this PR per the phase-1 plan):
- Completes the optional user_id bind at the read_context /
  read_node_memory / write_node_memory tool call sites (server.py),
  wiring get_user_id() into the already-hardened node_memory.py /
  nodes.py query functions from the prior WIP commit.
- Adds a regression test using the real (non-mocked) execute_read_context
  with conversation_id=None returning real root nodes, plus an RLS
  acceptance test proving a mismatched explicit user_id returns nothing
  even on an unscoped connection.

Matched pair with tether-premium feature/brief-1a-cascade-traverse-depth
(prompt_assembler.load_context_cascade kwarg N= -> traverse_depth=).
---
 tests/mcp/test_read_context_enforcement.py    | 118 ++++++-------
 .../mcp/test_write_node_memory_enforcement.py |   7 +-
 tests/tether_mcp/test_read_context_cascade.py | 142 ++++++++++-----
 tests/tether_mcp/test_server.py               |   9 +-
 tether_mcp/server.py                          |  32 ++--
 tether_mcp/tools/read_context.py              | 163 ++++++++----------
 tether_mcp/tools/read_node_memory.py          |   5 +
 tether_mcp/tools/write_node_memory.py         |   8 +-
 8 files changed, 272 insertions(+), 212 deletions(-)

diff --git a/tests/mcp/test_read_context_enforcement.py b/tests/mcp/test_read_context_enforcement.py
index 9d0f402b..b71b3f2b 100644
--- a/tests/mcp/test_read_context_enforcement.py
+++ b/tests/mcp/test_read_context_enforcement.py
@@ -1,12 +1,17 @@
-"""Tests for read_context v2 enforcement: conversation_id required.
+"""Tests for read_context v3 demotion: pure retrieval, no authorization.
 
-All tests mock the DB layer — no Postgres required.
+PermissionGate (interactive_agent_layer/permissions.py) is now the SOLE scope
+enforcer (design review §5.1). read_context no longer refuses anything itself:
+
+  1. conversation_id is OPTIONAL. When absent, read_context still returns real
+     data (no error envelope) — it simply cannot log a read credit.
+  2. There is no more out_of_scope error dict anywhere in read_context output.
+     Scope is judged upstream by the gate before the tool call is even made.
+  3. When conversation_id IS provided, read credits are still logged (for the
+     gate's read-before-write bookkeeping elsewhere), but this is bookkeeping,
+     not enforcement.
 
-Enforcement rules (v2):
-  1. conversation_id is required — returns {"error": "conversation_id_required", ...}
-     immediately if absent (no scope enforcement, no reads).
-  2. Scope envelope (out_of_scope errors) are already structured dicts — verified here.
-  3. When conversation_id is provided, execution proceeds normally.
+All tests mock the DB layer — no Postgres required.
 """
 from __future__ import annotations
 
@@ -19,14 +24,15 @@
 CONV_ID = "dddddddd-0000-0000-0000-000000000002"
 CURRENT_NODE = "eeeeeeee-0000-0000-0000-000000000003"
 
-# Patch targets for execute_read_context internals
+# Patch targets for execute_read_context internals.
+# Note: get_node_tree_distance is intentionally NOT here — the demoted
+# execute_read_context must not call it at all.
 PATCH_TARGETS = [
     "db.pg_queries.get_node",
     "db.pg_queries.get_node_by_path",
     "db.pg_queries.get_children",
     "db.pg_queries.node_memory.log_node_read",
     "db.pg_queries.node_memory.get_context_node_id_for_conversation",
-    "db.pg_queries.node_memory.get_node_tree_distance",
 ]
 
 
@@ -37,9 +43,7 @@ def conn():
 
 def _make_mocks() -> dict:
     mocks = {t.split(".")[-1]: AsyncMock() for t in PATCH_TARGETS}
-    # Default: conversation is linked to CURRENT_NODE, nodes are in scope
     mocks["get_context_node_id_for_conversation"].return_value = CURRENT_NODE
-    mocks["get_node_tree_distance"].return_value = 1  # within N=3
     mocks["get_node"].return_value = {
         "id": NODE_ID, "name": "Test Node",
         "section_types": [], "children_count": 0,
@@ -58,6 +62,18 @@ def _patch_all(mocks: dict):
         for t in PATCH_TARGETS:
             key = t.split(".")[-1]
             stack.enter_context(patch(t, mocks[key]))
+        # 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.
+        stack.enter_context(
+            patch(
+                "db.pg_queries.node_memory.get_node_tree_distance",
+                AsyncMock(side_effect=AssertionError(
+                    "execute_read_context must not call get_node_tree_distance "
+                    "(scope enforcement lives solely in PermissionGate)"
+                )),
+            )
+        )
         yield mocks
 
 
@@ -68,52 +84,51 @@ async def _run(conn, mocks, **kwargs):
 
 
 # ---------------------------------------------------------------------------
-# B-1: conversation_id required
+# conversation_id is optional — pure retrieval, not gated
 # ---------------------------------------------------------------------------
 
-class TestConversationIdRequired:
+class TestConversationIdOptional:
     @pytest.mark.asyncio
-    async def test_no_conversation_id_returns_error_dict(self, conn):
+    async def test_no_conversation_id_returns_list_not_error(self, conn):
         mocks = _make_mocks()
         result = await _run(conn, mocks, conversation_id=None)
-        assert isinstance(result, dict), (
-            "Without conversation_id, result must be a dict error envelope, got list"
+        assert isinstance(result, list), (
+            "Without conversation_id, read_context must still return real data, "
+            "not an error envelope"
         )
-        assert result.get("error") == "conversation_id_required"
 
     @pytest.mark.asyncio
-    async def test_no_conversation_id_error_has_message(self, conn):
+    async def test_no_conversation_id_with_node_ids_returns_data(self, conn):
         mocks = _make_mocks()
-        result = await _run(conn, mocks, conversation_id=None)
-        assert "message" in result
+        result = await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=None)
+        assert isinstance(result, list)
+        assert len(result) == 1
+        assert result[0].get("id") == NODE_ID
 
     @pytest.mark.asyncio
-    async def test_no_conversation_id_does_not_query_db(self, conn):
-        """No DB calls should happen when conversation_id is missing."""
+    async def test_no_conversation_id_with_paths_returns_data(self, conn):
         mocks = _make_mocks()
-        await _run(conn, mocks, conversation_id=None)
-        mocks["get_node"].assert_not_called()
-        mocks["get_node_by_path"].assert_not_called()
-        mocks["get_children"].assert_not_called()
-        mocks["get_context_node_id_for_conversation"].assert_not_called()
+        result = await _run(conn, mocks, paths=["Work/Alpha"], conversation_id=None)
+        assert isinstance(result, list)
+        assert len(result) == 1
 
     @pytest.mark.asyncio
-    async def test_no_conversation_id_with_paths_returns_error(self, conn):
+    async def test_no_conversation_id_does_not_log_read_credit(self, conn):
+        """No conversation context means nothing to log a credit against."""
         mocks = _make_mocks()
-        result = await _run(conn, mocks, paths=["Work/Alpha"], conversation_id=None)
-        assert isinstance(result, dict)
-        assert result.get("error") == "conversation_id_required"
+        await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=None)
+        mocks["log_node_read"].assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_no_conversation_id_with_node_ids_returns_error(self, conn):
+    async def test_no_conversation_id_does_not_resolve_scope_node(self, conn):
+        """No point resolving a context node for scope when nothing enforces scope here."""
         mocks = _make_mocks()
-        result = await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=None)
-        assert isinstance(result, dict)
-        assert result.get("error") == "conversation_id_required"
+        await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=None)
+        mocks["get_context_node_id_for_conversation"].assert_not_called()
 
 
 # ---------------------------------------------------------------------------
-# B-2: happy path — conversation_id provided
+# conversation_id present — read credit bookkeeping only, still no refusals
 # ---------------------------------------------------------------------------
 
 class TestReadContextWithConversationId:
@@ -131,30 +146,17 @@ async def test_with_node_id_and_conversation_id_returns_list(self, conn):
         assert len(result) == 1
 
     @pytest.mark.asyncio
-    async def test_scope_check_is_called_when_conversation_id_provided(self, conn):
+    async def test_with_conversation_id_logs_read_credit(self, conn):
         mocks = _make_mocks()
-        await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=CONV_ID, N=3)
-        mocks["get_context_node_id_for_conversation"].assert_called_once_with(conn, CONV_ID)
+        await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=CONV_ID)
+        mocks["log_node_read"].assert_called_once()
 
     @pytest.mark.asyncio
-    async def test_out_of_scope_node_returns_structured_error_dict(self, conn):
-        """Out-of-scope nodes return {error: 'out_of_scope', ...} dicts, not exceptions."""
+    async def test_no_out_of_scope_dict_ever_produced(self, conn):
+        """There is no distance/out-of-scope machinery left in read_context at all."""
         mocks = _make_mocks()
-        # Node is too far away — distance exceeds N
-        mocks["get_node_tree_distance"].return_value = None
-        result = await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=CONV_ID, N=3)
+        result = await _run(
+            conn, mocks, node_ids=[NODE_ID], conversation_id=CONV_ID, traverse_depth=1
+        )
         assert isinstance(result, list)
-        assert len(result) == 1
-        entry = result[0]
-        assert entry.get("error") == "out_of_scope"
-        assert "target" in entry
-        assert "message" in entry
-
-    @pytest.mark.asyncio
-    async def test_out_of_scope_does_not_raise_exception(self, conn):
-        """Scope violations must be structured responses, never unhandled exceptions."""
-        mocks = _make_mocks()
-        mocks["get_node_tree_distance"].return_value = None
-        # Should complete without raising
-        result = await _run(conn, mocks, node_ids=[NODE_ID], conversation_id=CONV_ID)
-        assert result is not None
+        assert result[0].get("error") != "out_of_scope"
diff --git a/tests/mcp/test_write_node_memory_enforcement.py b/tests/mcp/test_write_node_memory_enforcement.py
index 3bafb457..ef9b6724 100644
--- a/tests/mcp/test_write_node_memory_enforcement.py
+++ b/tests/mcp/test_write_node_memory_enforcement.py
@@ -140,7 +140,8 @@ async def test_delete_without_prior_read_returns_error(self, conn):
 
     @pytest.mark.asyncio
     async def test_has_read_check_uses_correct_args(self, conn):
-        """has_read_node_in_conversation is called with (conn, node_id, conversation_id)."""
+        """has_read_node_in_conversation is called with (conn, node_id, conversation_id),
+        plus the optional RLS-hardening user_id kwarg (None unless the caller binds it)."""
         mocks = _make_mocks(has_read=False)
         await _run(
             conn, mocks,
@@ -148,7 +149,9 @@ async def test_has_read_check_uses_correct_args(self, conn):
             mode="edit",
             conversation_id=CONV_ID,
         )
-        mocks["has_read_node_in_conversation"].assert_called_once_with(conn, NODE_ID, CONV_ID)
+        mocks["has_read_node_in_conversation"].assert_called_once_with(
+            conn, NODE_ID, CONV_ID, user_id=None
+        )
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/tether_mcp/test_read_context_cascade.py b/tests/tether_mcp/test_read_context_cascade.py
index 8abe7b32..680ffc79 100644
--- a/tests/tether_mcp/test_read_context_cascade.py
+++ b/tests/tether_mcp/test_read_context_cascade.py
@@ -1,6 +1,10 @@
-"""Unit tests for M-level cascade depth gating in execute_read_context.
+"""Unit tests for cascade traverse_depth gating in execute_read_context.
 
-Mocks all DB calls — no Postgres required.
+Mocked tests here cover pure cascade-depth mechanics — no Postgres required.
+The bottom of the file adds real (non-mocked) Postgres-backed tests per the
+brief-1a acceptance criteria: a regression test proving conversation_id=None
+still returns real root nodes, and an RLS-hardening test proving a mismatched
+explicit user_id returns nothing even on an unscoped connection.
 """
 from __future__ import annotations
 
@@ -19,14 +23,19 @@ def _node(node_id: str, name: str) -> dict:
 
 
 def _make_patches(children_map: dict, nodes: dict, conv_node_id=None):
-    """Return a list of patch context managers covering all DB calls in read_context."""
+    """Return a list of patch context managers covering all DB calls in read_context.
+
+    Note: get_node_tree_distance is intentionally NOT patched here (the demoted
+    execute_read_context no longer imports/calls it — scope enforcement lives
+    solely in PermissionGate now).
+    """
     async def fake_get_children(conn, parent_id):
         return children_map.get(str(parent_id) if parent_id else None, [])
 
     async def fake_get_node(conn, node_id):
         return nodes.get(str(node_id))
 
-    async def fake_get_node_by_path(conn, path):
+    async def fake_get_node_by_path(conn, path, *, user_id=None):
         for node in nodes.values():
             if node.get("path") == path or node.get("name") == path:
                 return node
@@ -43,10 +52,6 @@ async def fake_get_node_by_path(conn, path):
             new=AsyncMock(return_value=conv_node_id),
         ),
         patch("db.pg_queries.node_memory.log_node_read", new=AsyncMock()),
-        patch(
-            "db.pg_queries.node_memory.get_node_tree_distance",
-            new=AsyncMock(return_value=1),
-        ),
         patch("db.pg_queries.node_memory.get_node_summary", new=AsyncMock(return_value=None)),
         patch("tether_mcp.write_modes.format_cat_n", side_effect=lambda x: x),
         patch("tether_mcp.write_modes.line_count", return_value=1),
@@ -54,8 +59,9 @@ async def fake_get_node_by_path(conn, path):
 
 
 @pytest.mark.asyncio
-async def test_cascade_stops_at_N_depth_from_source():
-    """Children exactly N levels from source are included; their children are not fetched."""
+async def test_cascade_stops_at_traverse_depth_from_source():
+    """Children exactly traverse_depth levels from source are included; their
+    children are not fetched."""
     root = _node("root", "Root")
     child = _node("child", "Child")
     gc = _node("gc", "Grandchild")
@@ -71,7 +77,7 @@ async def test_cascade_stops_at_N_depth_from_source():
     @contextlib.asynccontextmanager
     async def apply_patches():
         with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patches[5], patches[6], patches[7], patches[8], patches[9], patches[10]:
+             patches[5], patches[6], patches[7], patches[8], patches[9]:
             yield
 
     async with apply_patches():
@@ -81,7 +87,7 @@ async def apply_patches():
             paths=["Root"],
             depth=-1,
             conversation_id="conv-1",
-            N=1,
+            traverse_depth=1,
         )
 
     assert isinstance(result, list)
@@ -90,15 +96,16 @@ async def apply_patches():
     assert "children" in root_r, "Root should have children key (depth > 0)"
     assert len(root_r["children"]) == 1
     child_r = root_r["children"][0]
-    # Grandchild is at depth 2 > N=1 — cascade must NOT recurse into it
+    # Grandchild is at depth 2 > traverse_depth=1 — cascade must NOT recurse into it
     assert "children" not in child_r or len(child_r.get("children", [])) == 0, (
-        "Children at cascade depth == N should not have their own children fetched"
+        "Children at cascade depth == traverse_depth should not have their own children fetched"
     )
 
 
 @pytest.mark.asyncio
-async def test_cascade_includes_nodes_up_to_and_including_N():
-    """Nodes at depth == N from source are in the result; they are just not expanded."""
+async def test_cascade_includes_nodes_up_to_and_including_traverse_depth():
+    """Nodes at depth == traverse_depth from source are in the result; they are
+    just not expanded."""
     root = _node("root", "Root")
     child = _node("child", "Child")
 
@@ -113,7 +120,7 @@ async def test_cascade_includes_nodes_up_to_and_including_N():
     @contextlib.asynccontextmanager
     async def apply_patches():
         with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patches[5], patches[6], patches[7], patches[8], patches[9], patches[10]:
+             patches[5], patches[6], patches[7], patches[8], patches[9]:
             yield
 
     async with apply_patches():
@@ -123,18 +130,18 @@ async def apply_patches():
             paths=["Root"],
             depth=-1,
             conversation_id="conv-1",
-            N=1,
+            traverse_depth=1,
         )
 
     root_r = result[0]
     assert "children" in root_r
     child_ids = [c.get("id") or c.get("name") for c in root_r["children"]]
-    assert "child" in child_ids, "Child at depth 1 == N should be included"
+    assert "child" in child_ids, "Child at depth 1 == traverse_depth should be included"
 
 
 @pytest.mark.asyncio
-async def test_cascade_unlimited_when_N_is_zero():
-    """N=0 means no M-gating — traverse all children (backward compat)."""
+async def test_cascade_unlimited_when_traverse_depth_is_zero():
+    """traverse_depth=0 means no cascade-gating — traverse all children (backward compat)."""
     root = _node("root", "Root")
     child = _node("child", "Child")
     gc = _node("gc", "Grandchild")
@@ -150,7 +157,7 @@ async def test_cascade_unlimited_when_N_is_zero():
     @contextlib.asynccontextmanager
     async def apply_patches():
         with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patches[5], patches[6], patches[7], patches[8], patches[9], patches[10]:
+             patches[5], patches[6], patches[7], patches[8], patches[9]:
             yield
 
     async with apply_patches():
@@ -160,19 +167,19 @@ async def apply_patches():
             paths=["Root"],
             depth=-1,
             conversation_id="conv-1",
-            N=0,
+            traverse_depth=0,
         )
 
     root_r = result[0]
     assert "children" in root_r
     child_r = root_r["children"][0]
-    # N=0 means no gating — grandchild should be fetched
-    assert "children" in child_r, "N=0 should not gate cascade — grandchild must be fetched"
+    # traverse_depth=0 means no gating — grandchild should be fetched
+    assert "children" in child_r, "traverse_depth=0 should not gate cascade — grandchild must be fetched"
 
 
 @pytest.mark.asyncio
 async def test_cascade_respects_depth_param_independently():
-    """depth=1 still limits traversal even when N > depth."""
+    """depth=1 still limits traversal even when traverse_depth > depth."""
     root = _node("root", "Root")
     child = _node("child", "Child")
     gc = _node("gc", "Grandchild")
@@ -188,7 +195,7 @@ async def test_cascade_respects_depth_param_independently():
     @contextlib.asynccontextmanager
     async def apply_patches():
         with patches[0], patches[1], patches[2], patches[3], patches[4], \
-             patches[5], patches[6], patches[7], patches[8], patches[9], patches[10]:
+             patches[5], patches[6], patches[7], patches[8], patches[9]:
             yield
 
     async with apply_patches():
@@ -198,7 +205,7 @@ async def apply_patches():
             paths=["Root"],
             depth=1,   # only 1 level
             conversation_id="conv-1",
-            N=5,       # N allows 5 levels, but depth caps at 1
+            traverse_depth=5,       # allows 5 levels, but depth caps at 1
         )
 
     root_r = result[0]
@@ -208,20 +215,73 @@ async def apply_patches():
     assert "children" not in child_r, "depth=1 should prevent grandchild from appearing"
 
 
+# ---------------------------------------------------------------------------
+# Real-Postgres regression + RLS-hardening acceptance tests (brief-1a)
+# ---------------------------------------------------------------------------
+# Uses tests/tether_mcp/conftest.py's `conn` fixture: real RLS-scoped
+# transactional connection, rolled back after each test. Skips automatically
+# without DATABASE_URL.
+
+import uuid as _uuid
+
+
 @pytest.mark.asyncio
-async def test_conversation_id_required():
-    """execute_read_context returns error dict when conversation_id is absent."""
-    conn = MagicMock()
+async def test_real_execute_read_context_no_conversation_id_returns_real_roots(conn):
+    """Regression test (the #12 lesson): no mocks. A real root context node
+    must come back from execute_read_context when conversation_id=None —
+    demotion must not silently return nothing or an error dict."""
+    from tests.tether_mcp.conftest import TEST_USER_ID
+
+    node_id = str(_uuid.uuid4())
+    await conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES ($1::uuid, $2::uuid, NULL, $3)",
+        node_id, TEST_USER_ID, "RealRootBrief1a",
+    )
 
-    async def fake_get_context_node_id(conn, conversation_id):
-        return None
+    from tether_mcp.tools.read_context import execute_read_context
+    result = await execute_read_context(conn, conversation_id=None)
 
-    with patch(
-        "db.pg_queries.node_memory.get_context_node_id_for_conversation",
-        new=AsyncMock(side_effect=fake_get_context_node_id),
-    ):
-        from tether_mcp.tools.read_context import execute_read_context
-        result = await execute_read_context(conn, conversation_id=None)
+    assert isinstance(result, list)
+    assert any(r.get("id") == node_id for r in result), (
+        "Real root node must be returned with conversation_id=None — "
+        "read_context is pure retrieval now, not gated on conversation context"
+    )
+
+
+@pytest.mark.asyncio
+async def test_rls_hardening_mismatched_user_id_returns_nothing(conn):
+    """Defense-in-depth: even on a connection where RLS/session GUC would
+    otherwise allow it, an explicit mismatched user_id must return nothing
+    from the hardened node_memory queries."""
+    from tests.tether_mcp.conftest import TEST_USER_ID
+    from db.pg_queries.node_memory import log_node_read, get_conversation_reads
+
+    conversation_id = str(_uuid.uuid4())
+    node_id = str(_uuid.uuid4())
+    other_user_id = str(_uuid.uuid4())
+    await conn.execute(
+        "INSERT INTO users (id, username, email, password_hash, is_admin) "
+        "VALUES ($1::uuid, 'brief1a_other', 'brief1a_other@example.com', 'x', false) "
+        "ON CONFLICT DO NOTHING",
+        other_user_id,
+    )
+    await conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES ($1::uuid, $2::uuid, NULL, $3)",
+        node_id, TEST_USER_ID, "RLSHardeningNode",
+    )
+    await conn.execute(
+        "INSERT INTO conversations (id, user_id, name) VALUES ($1::uuid, $2::uuid, $3)",
+        conversation_id, TEST_USER_ID, "brief1a RLS hardening test conversation",
+    )
+
+    await log_node_read(
+        conn, node_id, 4, conversation_id=conversation_id, user_id=TEST_USER_ID,
+    )
 
-    assert isinstance(result, dict)
-    assert result.get("error") == "conversation_id_required"
+    # Explicit mismatched user_id must see nothing, independent of the
+    # session's app.current_user_id GUC (which is set to TEST_USER_ID here —
+    # the point is the explicit bind overrides/ignores that for isolation).
+    reads = await get_conversation_reads(conn, conversation_id, user_id=other_user_id)
+    assert reads == []
diff --git a/tests/tether_mcp/test_server.py b/tests/tether_mcp/test_server.py
index e6b52d7f..39cdf7af 100644
--- a/tests/tether_mcp/test_server.py
+++ b/tests/tether_mcp/test_server.py
@@ -58,12 +58,13 @@ async def test_read_tasks_all(seeded):
 
 
 @pytest.mark.asyncio
-async def test_read_context_requires_conversation_id(seeded):
-    """v2: read_context without conversation_id returns a structured error, not a list."""
+async def test_read_context_without_conversation_id_returns_real_data(seeded):
+    """read_context is pure retrieval (brief-1a demotion): no conversation_id
+    required, no error envelope — real root nodes come back regardless."""
     from tether_mcp.server import read_context
     result = await read_context()
-    assert isinstance(result, dict), "Expected error dict when conversation_id is absent"
-    assert result.get("error") == "conversation_id_required"
+    assert isinstance(result, list), "read_context must return real data, not an error dict"
+    assert any(r.get("name") == "Work" for r in result)
 
 
 @pytest.mark.asyncio
diff --git a/tether_mcp/server.py b/tether_mcp/server.py
index 4255cb1f..bfdbc97d 100644
--- a/tether_mcp/server.py
+++ b/tether_mcp/server.py
@@ -123,26 +123,34 @@ async def read_context(
     include_tasks: bool = False,
     conversation_id: str = "",
     M: int = 4,
-    N: int = 3,
+    traverse_depth: int = 3,
     source: str = "sections",
 ) -> list:
     """Read context nodes. No params=roots. depth: 0=node only, 1=children, -1=full subtree.
     Section bodies in cat-n format (1-indexed line numbers with tabs).
 
-    conversation_id: Current conversation UUID. REQUIRED (v2) — returns
-        {error: 'conversation_id_required'} if absent.
-    M: Detail level for node data summary (1=title, 2=one-liner, 3=themes, 4=full sections).
-    N: Scope envelope — max tree-edges from conversation's context node. Nodes outside
-        N edges return {error: 'out_of_scope', target: ...} instead of data.
+    Pure retrieval — scope/authorization is judged upstream by PermissionGate
+    before this tool is called; this tool never refuses a request itself.
+
+    conversation_id: Current conversation UUID. Optional — when present, reads
+        are credit-logged for bookkeeping (e.g. write_node_memory's
+        read-before-write check).
+    M: Detail level requested for node data summary (1=title, 2=one-liner,
+        3=themes, 4=full sections) — subject to the gate's own judgment of
+        what detail is appropriate, not enforced here.
+    traverse_depth: Cascade cost bound — max tree-edges from the source node
+        before children stop being expanded. Not an authorization boundary.
     source: 'sections' (user-authored, default) | 'memory' (bot-authored) | 'both'.
     """
     from tether_mcp.tools.read_context import execute_read_context
     pool = await _get_pool()
-    async with pg.get_conn(pool, get_user_id()) as conn:
+    user_id = get_user_id()
+    async with pg.get_conn(pool, user_id) as conn:
         return await execute_read_context(
             conn, paths, node_ids, depth, include_sections, include_tasks,
             conversation_id=conversation_id or None,
-            M=M, N=N, source=source,
+            M=M, traverse_depth=traverse_depth, source=source,
+            user_id=user_id,
         )
 
 
@@ -188,13 +196,15 @@ async def read_node_memory(
     """
     from tether_mcp.tools.read_node_memory import execute_read_node_memory
     pool = await _get_pool()
-    async with pg.get_conn(pool, get_user_id()) as conn:
+    user_id = get_user_id()
+    async with pg.get_conn(pool, user_id) as conn:
         return await execute_read_node_memory(
             conn,
             node_id=node_id,
             title=title or None,
             M=M,
             conversation_id=conversation_id or None,
+            user_id=user_id,
         )
 
 
@@ -225,7 +235,8 @@ async def write_node_memory(
     """
     from tether_mcp.tools.write_node_memory import execute_write_node_memory
     pool = await _get_pool()
-    async with pg.get_conn(pool, get_user_id()) as conn:
+    user_id = get_user_id()
+    async with pg.get_conn(pool, user_id) as conn:
         return await execute_write_node_memory(
             conn,
             node_id=node_id,
@@ -235,6 +246,7 @@ async def write_node_memory(
             mode=mode,
             conversation_id=conversation_id or None,
             visible_to_user=visible_to_user,
+            user_id=user_id,
         )
 
 
diff --git a/tether_mcp/tools/read_context.py b/tether_mcp/tools/read_context.py
index dc749b42..edcb0d9c 100644
--- a/tether_mcp/tools/read_context.py
+++ b/tether_mcp/tools/read_context.py
@@ -1,17 +1,25 @@
 """read_context tool — batched reads with depth traversal, sections (cat -n), tasks,
-M-level summary, scope envelope enforcement, and source filtering.
-
-Params added/hardened in memory-context v2:
-  conversation_id — REQUIRED (v2). Returns {error: 'conversation_id_required'} if absent.
-                    Enables N scope envelope enforcement.
+M-level summary, and source filtering.
+
+read_context is PURE RETRIEVAL. It does not enforce scope or authorization —
+PermissionGate (interactive_agent_layer/permissions.py) is the sole enforcer
+(design review §5.1); by the time a read_context call reaches this module, the
+gate has already judged whether it should happen. This module never refuses a
+read and never returns an out_of_scope error dict.
+
+Params:
+  conversation_id — OPTIONAL. When present, each read logs a read-credit
+                    against it (bookkeeping only, e.g. for write_node_memory's
+                    read-before-write check). When absent, reads simply are
+                    not credited — retrieval still proceeds normally.
   M               — M-level detail for node data summary.
                     1=title only, 2=one-liner, 3=themes+abstract, 4=full (default: 4).
                     When M < last, returns from node_data_summary if cached;
                     falls back to node_sections at M=4.
-  N               — Scope envelope in tree-edges from current_node.
-                    Requests outside N edges return {error: 'out_of_scope'}.
-                    Note: children returned during depth traversal are also scope-checked;
-                    out-of-scope children are replaced with {error: 'out_of_scope', ...}.
+  traverse_depth  — Cost bound in tree-edges from current_node: how far the
+                    cascade descends before it stops expanding children.
+                    Not an authorization boundary — nodes beyond the bound are
+                    simply not expanded, no error dicts.
   source          — 'sections' | 'memory' | 'both' (default: 'sections').
                     'sections': return user-authored sections (origin='user').
                     'memory':   return bot-authored sections (origin='conversation_agent').
@@ -36,20 +44,19 @@ async def _build_node_response(
     M: int = 4,
     source: str = "sections",
     *,
-    current_node_id: str | None = None,
-    N: int = 3,
     conversation_id: str | None = None,
+    user_id: str | None = None,
+    traverse_depth: int = 3,
     _cascade_depth: int = 0,
 ) -> dict:
     """Build the response dict for a single node, optionally with children/sections/tasks.
 
-    When current_node_id is set, each child is scope-checked against N edges.
-    Out-of-scope children are represented as {error: 'out_of_scope', target: id}.
     _cascade_depth tracks levels descended from the read_context source node;
-    when it reaches N, _add_children will not recurse further.
+    when it reaches traverse_depth, _add_children will not recurse further
+    (a cost bound, not an authorization boundary).
     """
     from db.pg_queries import get_node, get_sections, get_node_tasks
-    from db.pg_queries.node_memory import get_node_summary, log_node_read
+    from db.pg_queries.node_memory import get_node_summary
     from tether_mcp.write_modes import format_cat_n, line_count
 
     # Ensure we have full node dict (with section_types and children_count)
@@ -77,9 +84,8 @@ async def _build_node_response(
                 if depth != 0:
                     await _add_children(
                         conn, result, node, depth, include_sections, include_tasks,
-                        M, source, current_node_id=current_node_id, N=N,
-                        conversation_id=conversation_id,
-                        _cascade_depth=_cascade_depth,
+                        M, source, conversation_id=conversation_id, user_id=user_id,
+                        traverse_depth=traverse_depth, _cascade_depth=_cascade_depth,
                     )
                 return result
 
@@ -112,13 +118,12 @@ async def _build_node_response(
     if include_tasks:
         result["tasks"] = await get_node_tasks(conn, node["id"])
 
-    # Children (recursive, with per-child scope check)
+    # Children (recursive)
     if depth != 0:
         await _add_children(
             conn, result, node, depth, include_sections, include_tasks,
-            M, source, current_node_id=current_node_id, N=N,
-            conversation_id=conversation_id,
-            _cascade_depth=_cascade_depth,
+            M, source, conversation_id=conversation_id, user_id=user_id,
+            traverse_depth=traverse_depth, _cascade_depth=_cascade_depth,
         )
 
     return result
@@ -134,21 +139,23 @@ async def _add_children(
     M: int,
     source: str,
     *,
-    current_node_id: str | None,
-    N: int,
     conversation_id: str | None,
+    user_id: str | None,
+    traverse_depth: int,
     _cascade_depth: int = 0,
 ) -> None:
-    """Fetch children and add them to result['children'], applying per-child scope check.
+    """Fetch children and add them to result['children'].
 
-    Enforces M-level cascade depth gating: when N > 0 and _cascade_depth >= N,
-    children are not fetched (the current node is at the scope boundary).
+    Enforces cascade cost gating only: when traverse_depth > 0 and
+    _cascade_depth >= traverse_depth, children are not fetched (a cost bound,
+    not an authorization boundary — no error dicts are produced).
     """
     from db.pg_queries import get_children
-    from db.pg_queries.node_memory import get_node_tree_distance, log_node_read
+    from db.pg_queries.node_memory import log_node_read
 
-    # M-level cascade gating: stop descending when we've reached N levels from source
-    if N > 0 and _cascade_depth >= N:
+    # Cost-bound cascade gating: stop descending once traverse_depth levels
+    # from the read_context source node have been reached.
+    if traverse_depth > 0 and _cascade_depth >= traverse_depth:
         return
 
     children = await get_children(conn, node["id"])
@@ -158,27 +165,14 @@ async def _add_children(
     for child in children:
         child_id = str(child["id"]) if not isinstance(child["id"], str) else child["id"]
 
-        # Scope check each child independently
-        if current_node_id and N > 0:
-            dist = await get_node_tree_distance(conn, current_node_id, child_id, N)
-            if dist is None:
-                result["children"].append({
-                    "error": "out_of_scope",
-                    "target": child_id,
-                    "message": (
-                        f"Node {child_id} is more than {N} tree-edges from "
-                        f"conversation context node {current_node_id}."
-                    ),
-                })
-                continue
-
-        # Log read credit for in-scope children
+        # Log read credit for children read during cascade traversal
         if conversation_id:
             try:
                 await log_node_read(
                     conn, child_id, M,
                     conversation_id=conversation_id,
                     title=child.get("name"),
+                    user_id=user_id,
                 )
             except Exception:
                 pass
@@ -186,8 +180,8 @@ async def _add_children(
         result["children"].append(
             await _build_node_response(
                 conn, child, next_depth, include_sections, include_tasks, M, source,
-                current_node_id=current_node_id, N=N, conversation_id=conversation_id,
-                _cascade_depth=_cascade_depth + 1,
+                conversation_id=conversation_id, user_id=user_id,
+                traverse_depth=traverse_depth, _cascade_depth=_cascade_depth + 1,
             )
         )
 
@@ -201,11 +195,14 @@ async def execute_read_context(
     include_tasks: bool = False,
     conversation_id: str | None = None,
     M: int = 4,
-    N: int = 3,
+    traverse_depth: int = 3,
     source: str = "sections",
-) -> list | dict:
+    user_id: str | None = None,
+) -> list:
     """Fetch context nodes with optional depth traversal, section content, and linked tasks.
 
+    Pure retrieval — no authorization is performed here (see module docstring).
+
     Args:
         conn: asyncpg connection (user-scoped via RLS).
         paths: List of slash-separated paths like ["Projects/Tether"]. Resolved to nodes.
@@ -217,86 +214,62 @@ async def execute_read_context(
         include_sections: If True, add "sections" dict grouped by section_type.
             Each section entry has {name, body (cat-n format), line_count, origin}.
         include_tasks: If True, add "tasks" list from get_node_tasks.
-        conversation_id: Current conversation UUID. Required (v2) — returns
-            {error: 'conversation_id_required'} if absent.
+        conversation_id: Current conversation UUID. Optional — when present,
+            reads are credit-logged for bookkeeping (e.g. write_node_memory's
+            read-before-write check); when absent, retrieval proceeds
+            identically, just without logging.
         M: M-level detail for node data (1=title, 2=one-liner, 3=themes, 4=full).
             When M < 4, returns from node_data_summary cache if available;
             falls back to node_sections at M=4.
-        N: Scope envelope — max tree-edges from conversation's current_node.
-            Requests outside N edges return {error: 'out_of_scope', ...}.
-            Children during depth traversal are also scope-checked.
+        traverse_depth: Cascade cost bound — max tree-edges from the source
+            node before children stop being expanded. Not an authorization
+            boundary.
         source: 'sections' (default, user-authored) | 'memory' (bot-authored) | 'both'.
+        user_id: RLS hardening — explicit caller-supplied user_id, bound
+            directly on read-credit inserts instead of relying solely on the
+            session GUC. Optional, backward compatible.
 
     Returns:
-        List of node dicts (or error dicts for out-of-scope entries).
-        If no paths and no node_ids, returns root nodes.
-        {error: 'conversation_id_required', message: str} if conversation_id is absent.
+        List of node dicts. If no paths and no node_ids, returns root nodes.
     """
     from db.pg_queries import get_node, get_node_by_path, get_children
-    from db.pg_queries.node_memory import (
-        log_node_read,
-        get_context_node_id_for_conversation,
-        get_node_tree_distance,
-    )
-
-    # v2: conversation_id is required
-    if not conversation_id:
-        return {
-            "error": "conversation_id_required",
-            "message": "conversation_id is required for read_context in v2.",
-        }
-
-    # Resolve conversation scope (current_node_id may be None if conversation is unlinked)
-    current_node_id = await get_context_node_id_for_conversation(conn, conversation_id)
-
-    async def _fetch_and_check(node: dict) -> dict:
-        """Apply scope check, log read, build response."""
+    from db.pg_queries.node_memory import log_node_read
+
+    async def _fetch_and_log(node: dict) -> dict:
+        """Log read credit (if conversation context present), build response."""
         node_id = str(node["id"]) if not isinstance(node["id"], str) else node["id"]
 
-        # Scope envelope check for the top-level requested node
-        if current_node_id and N > 0:
-            dist = await get_node_tree_distance(conn, current_node_id, node_id, N)
-            if dist is None:
-                return {
-                    "error": "out_of_scope",
-                    "target": node_id,
-                    "message": (
-                        f"Node {node_id} is more than {N} tree-edges from "
-                        f"conversation context node {current_node_id}. "
-                        "Request permission via the conversation to access it."
-                    ),
-                }
-
-        # Log read credit for the top-level node
         if conversation_id:
             try:
                 await log_node_read(
                     conn, node_id, M,
                     conversation_id=conversation_id,
                     title=node.get("name"),
+                    user_id=user_id,
                 )
             except Exception:
                 pass
 
         return await _build_node_response(
             conn, node, depth, include_sections, include_tasks, M, source,
-            current_node_id=current_node_id, N=N, conversation_id=conversation_id,
+            conversation_id=conversation_id, user_id=user_id,
+            traverse_depth=traverse_depth,
         )
 
     # No args → return root nodes
     if not paths and not node_ids:
         roots = await get_children(conn, parent_id=None)
-        return [await _fetch_and_check(root) for root in roots]
+        return [await _fetch_and_log(root) for root in roots]
 
     results = []
 
     # Resolve paths
     for path in (paths or []):
-        node = await get_node_by_path(conn, path)
+        node = await get_node_by_path(conn, path, user_id=user_id)
         if node is None:
             results.append(None)
         else:
-            results.append(await _fetch_and_check(node))
+            results.append(await _fetch_and_log(node))
 
     # Resolve node_ids
     for nid in (node_ids or []):
@@ -304,6 +277,6 @@ async def _fetch_and_check(node: dict) -> dict:
         if node is None:
             results.append(None)
         else:
-            results.append(await _fetch_and_check(node))
+            results.append(await _fetch_and_log(node))
 
     return results
diff --git a/tether_mcp/tools/read_node_memory.py b/tether_mcp/tools/read_node_memory.py
index ab303551..3ceb9b35 100644
--- a/tether_mcp/tools/read_node_memory.py
+++ b/tether_mcp/tools/read_node_memory.py
@@ -21,6 +21,7 @@ async def execute_read_node_memory(
     M: int = 4,
     *,
     conversation_id: str | None = None,
+    user_id: str | None = None,
 ) -> dict:
     """Return bot-authored sections for a context node.
 
@@ -30,6 +31,9 @@ async def execute_read_node_memory(
         title:           Optional section name filter.
         M:               Detail level — 1 (names only), 2 (preview), 3 (truncated), 4 (full).
         conversation_id: Current conversation UUID (for read-credit logging).
+        user_id:         RLS hardening — explicit caller-supplied user_id, bound
+                         directly on the read-credit insert. Optional, backward
+                         compatible.
 
     Returns:
         {node_id, sections: [{section_type, name, body?, preview?, origin}]}
@@ -87,6 +91,7 @@ def format_body(body: str | None) -> str | None:
                 conn, node_id, M,
                 conversation_id=conversation_id,
                 title=node.get("name"),
+                user_id=user_id,
             )
         except Exception:
             pass
diff --git a/tether_mcp/tools/write_node_memory.py b/tether_mcp/tools/write_node_memory.py
index 27213f2b..4e22992a 100644
--- a/tether_mcp/tools/write_node_memory.py
+++ b/tether_mcp/tools/write_node_memory.py
@@ -32,6 +32,7 @@ async def execute_write_node_memory(
     *,
     conversation_id: str | None = None,
     visible_to_user: bool = True,
+    user_id: str | None = None,
 ) -> dict:
     """Write a bot-authored section to a context node.
 
@@ -71,7 +72,9 @@ async def execute_write_node_memory(
         }
 
     # v2: a read of this node must exist in node_read_log for this conversation
-    has_read = await has_read_node_in_conversation(conn, node_id, conversation_id)
+    has_read = await has_read_node_in_conversation(
+        conn, node_id, conversation_id, user_id=user_id
+    )
     if not has_read:
         return {
             "error": "read_before_write_required",
@@ -104,7 +107,8 @@ async def execute_write_node_memory(
     # Log write as a read credit so future reads know this conversation touched the node
     try:
         await log_node_read(
-            conn, node_id, 999, conversation_id=conversation_id, title=title
+            conn, node_id, 999, conversation_id=conversation_id, title=title,
+            user_id=user_id,
         )
     except Exception:
         pass  # don't fail the write if log fails

From b18062b94e7be0128d3805a99c55cd05958d3893 Mon Sep 17 00:00:00 2001
From: Jason Lunder <jlunder@zagmail.gonzaga.edu>
Date: Sat, 4 Jul 2026 18:57:35 -0700
Subject: [PATCH 3/3] harden get_node with the same optional user_id bind as
 get_node_by_path
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Code review of the read_context demotion commit caught an inconsistency:
get_node_by_path was hardened with an optional user_id explicit-bind
param, but get_node (used by read_context's node_ids branch and the
children section_types/children_count backfill in _build_node_response)
was not touched at all — leaving id-based lookups one RLS layer weaker
than path-based ones within the same tool. This adds the same optional
user_id param to get_node and wires it through read_context.py's two
call sites plus read_node_memory.py's node lookup.

Also corrects two RLS-hardening test docstrings that overstated what
they verify: the local dev Postgres role connects BYPASSRLS (same root
cause as the pre-existing tests/db/test_rls.py::
test_app_db_role_is_not_superuser failure), so "RLS denies without an
explicit bind" cannot be exercised in this sandbox. What the new tests
(tests/db/test_pg_nodes.py) and the corrected existing test actually
verify — meaningfully, independent of RLS/role privileges — is that the
explicit user_id bind is a real filter: a mismatched user_id never
resolves another user's node, and a matching one does.
---
 db/pg_queries/nodes.py                        |  26 ++++-
 tests/db/test_pg_nodes.py                     | 104 +++++++++++++++++-
 tests/tether_mcp/test_read_context_cascade.py |  14 ++-
 tether_mcp/tools/read_context.py              |   4 +-
 tether_mcp/tools/read_node_memory.py          |   2 +-
 5 files changed, 138 insertions(+), 12 deletions(-)

diff --git a/db/pg_queries/nodes.py b/db/pg_queries/nodes.py
index 4a7a8680..6cfe29d7 100644
--- a/db/pg_queries/nodes.py
+++ b/db/pg_queries/nodes.py
@@ -47,10 +47,28 @@ async def create_node(
     return _node(row)
 
 
-async def get_node(conn: asyncpg.Connection, node_id: str) -> dict | None:
-    row = await conn.fetchrow(
-        "SELECT * FROM context_nodes WHERE id = $1", _uuid.UUID(node_id)
-    )
+async def get_node(
+    conn: asyncpg.Connection,
+    node_id: str,
+    *,
+    user_id: str | None = None,
+) -> dict | None:
+    """Fetch a single context node by id.
+
+    user_id: RLS hardening — when provided, binds this value directly as an
+    explicit `AND user_id = $N` filter instead of relying solely on
+    RLS/the session GUC. When None (default), behavior is unchanged
+    (RLS-only), backward compatible.
+    """
+    if user_id is not None:
+        row = await conn.fetchrow(
+            "SELECT * FROM context_nodes WHERE id = $1 AND user_id = $2::uuid",
+            _uuid.UUID(node_id), _uuid.UUID(user_id),
+        )
+    else:
+        row = await conn.fetchrow(
+            "SELECT * FROM context_nodes WHERE id = $1", _uuid.UUID(node_id)
+        )
     if not row:
         return None
     d = _node(row)
diff --git a/tests/db/test_pg_nodes.py b/tests/db/test_pg_nodes.py
index 9123ac64..ab8482fd 100644
--- a/tests/db/test_pg_nodes.py
+++ b/tests/db/test_pg_nodes.py
@@ -1,7 +1,7 @@
 """Tests for db/pg_queries/nodes.py — tree traversal, move, cycle detection."""
 import pytest
 
-from tests.db.pg_conftest import conn, TEST_USER_ID  # noqa: F401
+from tests.db.pg_conftest import conn, auth_conn, TEST_USER_ID  # noqa: F401
 from db.pg_queries.nodes import (
     create_node, get_node, get_node_by_path, get_children,
     ensure_node_path, get_all_node_paths, get_subtree,
@@ -111,3 +111,105 @@ async def test_get_all_node_paths(conn):
     await ensure_node_path(conn, "AllPaths/Sub/Leaf")
     paths = await get_all_node_paths(conn)
     assert any("AllPaths" in p for p in paths)
+
+
+# ---------------------------------------------------------------------------
+# RLS hardening (0e addendum / brief-1a) — real unscoped-connection tests.
+#
+# `auth_conn` (tests/db/pg_conftest.py) never sets `app.current_user_id`, so
+# these exercise the "defense in depth on an unscoped connection" case the
+# hardening addendum targets — unlike the `conn` fixture, which always has
+# the GUC set and so can never distinguish "explicit bind" from "GUC
+# happened to match".
+#
+# NOTE on local sandbox limits: the local dev Postgres role connects with
+# BYPASSRLS (confirmed: `rolbypassrls=True` for the `tether` role used by
+# DATABASE_URL here — see the pre-existing, out-of-scope failure of
+# tests/db/test_rls.py::test_app_db_role_is_not_superuser, which asserts
+# prod's app role is NOT superuser/bypassrls). That means RLS itself cannot
+# be exercised as a *deny* mechanism in this sandbox — a query with no
+# explicit user_id and no GUC will still see the row here, whereas in prod
+# (non-bypassing app role) RLS would deny it. What CAN be verified here,
+# independent of RLS/role privileges, is the actual code path this
+# hardening adds: an explicit user_id bind is a real SQL filter, so a
+# mismatched user_id must never resolve someone else's node, and a matching
+# explicit user_id must resolve it. That's what these tests assert.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_get_node_by_path_explicit_user_id_resolves_own_node(auth_conn):
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES (gen_random_uuid(), $1::uuid, NULL, $2)",
+        uuid.UUID(TEST_USER_ID), "ExplicitBindRoot",
+    )
+    found = await get_node_by_path(auth_conn, "ExplicitBindRoot", user_id=TEST_USER_ID)
+    assert found is not None
+    assert found["name"] == "ExplicitBindRoot"
+
+
+@pytest.mark.asyncio
+async def test_get_node_by_path_mismatched_user_id_returns_none(auth_conn):
+    """Explicit user_id binding must actually filter — a mismatched user_id
+    must not resolve a node that belongs to someone else."""
+    other_user_id = str(uuid.uuid4())
+    await auth_conn.execute(
+        "INSERT INTO users (id, username, email, password_hash, is_admin) "
+        "VALUES ($1::uuid, 'nodes_other', 'nodes_other@example.com', 'x', false) "
+        "ON CONFLICT DO NOTHING",
+        other_user_id,
+    )
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES (gen_random_uuid(), $1::uuid, NULL, $2)",
+        uuid.UUID(TEST_USER_ID), "MismatchRoot",
+    )
+    result = await get_node_by_path(auth_conn, "MismatchRoot", user_id=other_user_id)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_get_node_explicit_user_id_resolves_own_node(auth_conn):
+    """get_node gains the same optional user_id explicit-bind hardening as
+    get_node_by_path — brief-1a review finding: id-based lookups (used by
+    read_context's node_ids branch and the children section_types/
+    children_count backfill) previously had no user_id param at all."""
+    node_id = str(uuid.uuid4())
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES ($1::uuid, $2::uuid, NULL, $3)",
+        uuid.UUID(node_id), uuid.UUID(TEST_USER_ID), "ExplicitBindGetNode",
+    )
+    found = await get_node(auth_conn, node_id, user_id=TEST_USER_ID)
+    assert found is not None
+    assert found["name"] == "ExplicitBindGetNode"
+
+
+@pytest.mark.asyncio
+async def test_get_node_mismatched_user_id_returns_none(auth_conn):
+    node_id = str(uuid.uuid4())
+    other_user_id = str(uuid.uuid4())
+    await auth_conn.execute(
+        "INSERT INTO users (id, username, email, password_hash, is_admin) "
+        "VALUES ($1::uuid, 'nodes_other2', 'nodes_other2@example.com', 'x', false) "
+        "ON CONFLICT DO NOTHING",
+        other_user_id,
+    )
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES ($1::uuid, $2::uuid, NULL, $3)",
+        uuid.UUID(node_id), uuid.UUID(TEST_USER_ID), "MismatchGetNode",
+    )
+    result = await get_node(auth_conn, node_id, user_id=other_user_id)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_get_node_user_id_is_optional_backward_compatible(conn):
+    """user_id=None (default) preserves old RLS-only behavior on a
+    GUC-scoped connection — no regression for existing callers."""
+    node = await create_node(conn, name="BackCompatGetNode", node_type="context")
+    found = await get_node(conn, node["id"])
+    assert found is not None
+    assert found["id"] == node["id"]
diff --git a/tests/tether_mcp/test_read_context_cascade.py b/tests/tether_mcp/test_read_context_cascade.py
index 680ffc79..c5e94a0c 100644
--- a/tests/tether_mcp/test_read_context_cascade.py
+++ b/tests/tether_mcp/test_read_context_cascade.py
@@ -32,7 +32,7 @@ def _make_patches(children_map: dict, nodes: dict, conv_node_id=None):
     async def fake_get_children(conn, parent_id):
         return children_map.get(str(parent_id) if parent_id else None, [])
 
-    async def fake_get_node(conn, node_id):
+    async def fake_get_node(conn, node_id, *, user_id=None):
         return nodes.get(str(node_id))
 
     async def fake_get_node_by_path(conn, path, *, user_id=None):
@@ -251,9 +251,15 @@ async def test_real_execute_read_context_no_conversation_id_returns_real_roots(c
 
 @pytest.mark.asyncio
 async def test_rls_hardening_mismatched_user_id_returns_nothing(conn):
-    """Defense-in-depth: even on a connection where RLS/session GUC would
-    otherwise allow it, an explicit mismatched user_id must return nothing
-    from the hardened node_memory queries."""
+    """Defense-in-depth: the explicit user_id bind is a real filter, not a
+    no-op — a mismatched user_id must return nothing from the hardened
+    node_memory queries even though this connection's session GUC
+    (app.current_user_id) is set to TEST_USER_ID and would otherwise permit
+    it via RLS alone. (This test uses the GUC-scoped `conn` fixture, not an
+    unscoped connection — see tests/db/test_pg_nodes.py's RLS-hardening
+    block for the unscoped-connection case and its sandbox caveat: the
+    local dev Postgres role is BYPASSRLS, so RLS-as-deny can't be exercised
+    directly here; the explicit-bind code path can be, and is, above.)"""
     from tests.tether_mcp.conftest import TEST_USER_ID
     from db.pg_queries.node_memory import log_node_read, get_conversation_reads
 
diff --git a/tether_mcp/tools/read_context.py b/tether_mcp/tools/read_context.py
index edcb0d9c..acc40431 100644
--- a/tether_mcp/tools/read_context.py
+++ b/tether_mcp/tools/read_context.py
@@ -61,7 +61,7 @@ async def _build_node_response(
 
     # Ensure we have full node dict (with section_types and children_count)
     if "section_types" not in node or "children_count" not in node:
-        full = await get_node(conn, node["id"])
+        full = await get_node(conn, node["id"], user_id=user_id)
         if full is None:
             return node
         node = full
@@ -273,7 +273,7 @@ async def _fetch_and_log(node: dict) -> dict:
 
     # Resolve node_ids
     for nid in (node_ids or []):
-        node = await get_node(conn, nid)
+        node = await get_node(conn, nid, user_id=user_id)
         if node is None:
             results.append(None)
         else:
diff --git a/tether_mcp/tools/read_node_memory.py b/tether_mcp/tools/read_node_memory.py
index 3ceb9b35..ebbf944c 100644
--- a/tether_mcp/tools/read_node_memory.py
+++ b/tether_mcp/tools/read_node_memory.py
@@ -44,7 +44,7 @@ async def execute_read_node_memory(
     from db.pg_queries.sections import get_sections
 
     # Verify node exists
-    node = await get_node(conn, node_id)
+    node = await get_node(conn, node_id, user_id=user_id)
     if node is None:
         return {"error": "node_not_found", "node_id": node_id}