fix: workspace message tools deny-all under rebased session scoping

The #882 rebase changed the unscoped-observer contract from falsy to
'observer is None': resolve_session_scope looked up the workspace
executor's observer='' sentinel as a real peer with no session
memberships and denied every workspace-flat message read (search, grep,
date-range, temporal, observation context) whenever no session was
pinned — the primary workspace-chat shape. Normalize the sentinel to
None at the five read-handler crud boundaries and add regression tests
that run the tools unpinned (verified to fail without the fix).

Also from review:
- wrap the workspace prefetch in the same degrade-to-None protection
  the base agent has (an overview query error no longer 500s the
  request or kills the SSE stream after headers)
- thread session_allowlist through create_workspace_tool_executor so
  the agent-level allowlist seam is honored end to end when scopes
  (#897) wire it up; allowlisted grep is covered by a test
- deterministic name tie-break in get_active_peers ordering

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/crud/workspace.py

@@ -653,11 +653,15 @@ async def get_active_peers(
         .where(models.Peer.workspace_name == workspace_name)
     )
 
+    # Peer name as secondary key so ties (notably all-NULL activity in young
+    # workspaces) return a stable order across calls.
     if sort_by == "message_count":
-        stmt = stmt.order_by(func.coalesce(subq.c.msg_count, 0).desc())
+        stmt = stmt.order_by(
+            func.coalesce(subq.c.msg_count, 0).desc(), models.Peer.name
+        )
     else:
         # Default: recent_activity — peers with most recent messages first
-        stmt = stmt.order_by(subq.c.last_msg_at.desc().nulls_last())
+        stmt = stmt.order_by(subq.c.last_msg_at.desc().nulls_last(), models.Peer.name)
 
     stmt = stmt.limit(limit)
 

src/dialectic/workspace.py

@@ -78,23 +78,32 @@ class WorkspaceDialecticAgent(DialecticAgent):
         is known about them at a glance.
         """
         _ = query
-        async with tracked_db("dialectic.workspace_prefetch", read_only=True) as db:
-            stats = await crud.get_workspace_stats(db, self.workspace_name)
-            if stats.peer_count == 0:
-                return None
-            peers = await crud.get_active_peers(
-                db, self.workspace_name, limit=_PREFETCH_ACTIVE_PEERS
-            )
-            cards: dict[str, list[str]] = {}
-            for peer in peers:
-                card = await crud.get_peer_card(
-                    db,
-                    workspace_name=self.workspace_name,
-                    observer=peer.name,
-                    observed=peer.name,
+        # Like the base agent, prefetch failure degrades to no prefetched
+        # block rather than failing the whole request (the caller in
+        # _prepare_query does not guard this).
+        try:
+            async with tracked_db(
+                "dialectic.workspace_prefetch", read_only=True
+            ) as db:
+                stats = await crud.get_workspace_stats(db, self.workspace_name)
+                if stats.peer_count == 0:
+                    return None
+                peers = await crud.get_active_peers(
+                    db, self.workspace_name, limit=_PREFETCH_ACTIVE_PEERS
                 )
-                if card:
-                    cards[peer.name] = card
+                cards: dict[str, list[str]] = {}
+                for peer in peers:
+                    card = await crud.get_peer_card(
+                        db,
+                        workspace_name=self.workspace_name,
+                        observer=peer.name,
+                        observed=peer.name,
+                    )
+                    if card:
+                        cards[peer.name] = card
+        except Exception as e:
+            logger.warning(f"Failed to prefetch workspace overview: {e}")
+            return None
 
         lines: list[str] = [
             f"Peers: {stats.peer_count}",
@@ -145,6 +154,7 @@ class WorkspaceDialecticAgent(DialecticAgent):
         return await create_workspace_tool_executor(
             workspace_name=self.workspace_name,
             session_name=self.session_name,
+            session_allowlist=self.session_allowlist,
             history_token_limit=settings.DIALECTIC.HISTORY_TOKEN_LIMIT,
             run_id=self._run_id,
             agent_type="workspace_dialectic",

src/utils/agent_tools.py

@@ -2000,7 +2000,7 @@ async def _handle_get_observation_context(
             workspace_name=ctx.workspace_name,
             session_name=ctx.session_name,
             message_ids=tool_input["message_ids"],
-            observer=ctx.observer,
+            observer=ctx.observer or None,
             session_allowlist=ctx.session_allowlist,
         )
         if not messages:
@@ -2043,7 +2043,7 @@ async def _handle_search_messages(
         limit=limit,
         context_window=2,
         embedding=query_embedding,
-        observer=ctx.observer,
+        observer=ctx.observer or None,
         session_allowlist=ctx.session_allowlist,
     )
     search_meta: dict[str, Any] = {
@@ -2080,7 +2080,7 @@ async def _handle_grep_messages(
         text=text,
         limit=limit,
         context_window=context_window,
-        observer=ctx.observer,
+        observer=ctx.observer or None,
         session_allowlist=ctx.session_allowlist,
     )
     if not snippets:
@@ -2145,7 +2145,7 @@ async def _handle_get_messages_by_date_range(
             before_date=before_date,
             limit=limit,
             order=order,
-            observer=ctx.observer,
+            observer=ctx.observer or None,
             session_allowlist=ctx.session_allowlist,
         )
         msg_count = len(messages)
@@ -2221,7 +2221,7 @@ async def _handle_search_messages_temporal(
         context_window=context_window,
         session_allowlist=ctx.session_allowlist,
         embedding=query_embedding,
-        observer=ctx.observer,
+        observer=ctx.observer or None,
     )
     date_filter: list[str] = []
     if after_date_str:
@@ -2902,8 +2902,11 @@ def _estimate_tokens_safe(text: str | None) -> int | None:
 # The workspace agent is not bound to an (observer, observed) pair. Handlers
 # that need a pair take it from tool_input (the agent routes first, then
 # supplies the pair); the rest are workspace-scoped reads. Message-search
-# fallthrough handlers run with observer="" which crud treats as "no
-# perspective scoping" -- correct for a workspace-level read.
+# fallthrough handlers run with observer="" and normalize it to None at the
+# crud boundary (`ctx.observer or None`) -- None means "no perspective
+# scoping", which is correct for a workspace-level read. The empty string
+# must never reach resolve_session_scope: it would be looked up as a real
+# peer with no session memberships and deny all results.
 # ---------------------------------------------------------------------------
 
 
@@ -3024,6 +3027,7 @@ def _workspace_handler_resolver(tool_name: str) -> Any:
 async def create_workspace_tool_executor(
     workspace_name: str,
     session_name: str | None = None,
+    session_allowlist: list[str] | None = None,
     history_token_limit: int = 8192,
     run_id: str | None = None,
     agent_type: str | None = None,
@@ -3032,14 +3036,18 @@ async def create_workspace_tool_executor(
     """Tool executor for workspace-level operations (no bound peer pair).
 
     Reuses create_tool_executor's telemetry/error plumbing via the
-    handler_resolver seam; observer/observed are empty-string sentinels only
-    ever seen by handlers in _WORKSPACE_SAFE_FALLTHROUGH_TOOLS.
+    handler_resolver seam. observer/observed are empty-string sentinels only
+    ever seen by handlers in _WORKSPACE_SAFE_FALLTHROUGH_TOOLS, which
+    normalize them to None before hitting crud (None means "no perspective
+    scoping"; an empty string would read as a real peer with no sessions and
+    deny everything).
     """
     return await create_tool_executor(
         workspace_name=workspace_name,
         observer="",
         observed="",
         session_name=session_name,
+        session_allowlist=session_allowlist,
         include_observation_ids=True,
         history_token_limit=history_token_limit,
         run_id=run_id,

tests/test_workspace_chat.py

@@ -1196,3 +1196,88 @@ class TestFormatDocumentsWithAttribution:
 
         assert "coffee" in result.lower() or "programming" in result.lower()
         assert "remotely" in result.lower() or "works" in result.lower()
+
+
+# =============================================================================
+# Regression: workspace-flat message visibility without a pinned session
+# =============================================================================
+
+
+@pytest.mark.asyncio
+class TestWorkspaceMessageToolsUnpinned:
+    """The workspace executor's observer='' sentinel must read as
+    'no perspective scoping' (None) at the crud boundary. Under #882's
+    resolve_session_scope, an empty STRING is looked up as a real peer with
+    no session memberships and denies every result — so these tests run the
+    message tools with NO session_name, the primary workspace-chat shape."""
+
+    async def test_grep_messages_finds_content_without_session(
+        self,
+        db_session: AsyncSession,  # pyright: ignore[reportUnusedParameter]
+        workspace_test_data: Any,
+    ):
+        workspace, *_ = workspace_test_data
+
+        executor = await create_workspace_tool_executor(
+            workspace_name=workspace.name,
+        )
+        result = await executor("grep_messages", {"text": "Test message"})
+
+        assert isinstance(result, str)
+        assert "No messages found" not in result
+        assert "Test message" in result
+
+    async def test_date_range_finds_content_without_session(
+        self,
+        db_session: AsyncSession,  # pyright: ignore[reportUnusedParameter]
+        workspace_test_data: Any,
+    ):
+        workspace, *_ = workspace_test_data
+
+        executor = await create_workspace_tool_executor(
+            workspace_name=workspace.name,
+        )
+        result = await executor("get_messages_by_date_range", {"limit": 10})
+
+        assert isinstance(result, str)
+        assert "Found" in result
+        assert "No messages found" not in result
+
+    async def test_session_allowlist_is_honored_when_set(
+        self,
+        db_session: AsyncSession,  # pyright: ignore[reportUnusedParameter]
+        workspace_test_data: Any,
+    ):
+        """When a caller ever passes an allowlist (scopes work, #897), the
+        fallthrough message tools must intersect with it — an allowlist
+        naming no real session yields no results."""
+        workspace, *_ = workspace_test_data
+
+        executor = await create_workspace_tool_executor(
+            workspace_name=workspace.name,
+            session_allowlist=["no-such-session"],
+        )
+        result = await executor("grep_messages", {"text": "Test message"})
+
+        assert isinstance(result, str)
+        assert "No messages found" in result
+
+
+@pytest.mark.asyncio
+async def test_workspace_prefetch_failure_degrades_to_none(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Prefetch errors must not fail the request (parity with the base
+    agent's try/except): the agent proceeds with no prefetched block."""
+    from src.dialectic.workspace import WorkspaceDialecticAgent
+
+    async def boom(*args: Any, **kwargs: Any) -> Any:
+        _ = (args, kwargs)
+        raise RuntimeError("stats query exploded")
+
+    monkeypatch.setattr("src.dialectic.workspace.crud.get_workspace_stats", boom)
+
+    agent = WorkspaceDialecticAgent(workspace_name="w")
+    result = await agent._prefetch_relevant_observations("q")  # pyright: ignore[reportPrivateUsage]
+
+    assert result is None