fix(contextual): narrow tool palette + forbid legacy read tools

Smoke trace 0b46841484ba7d024ed9f8d5ac8b1df0 showed the agent
defaulting to list_projects + get_project for a 'summarize
project Nexus' query, returning a shallow row without aiSummary
or tasks/notes. The legacy read tools were exposed via
*PROJECT_TOOLS / *TASK_TOOLS spreading.

Now _contextual_tools exposes exactly:
- get_page_details (sole read; supports per-entity + list views)
- create_task, update_task
- create_note
- create_timeline

Prompt rule 2 explicitly forbids the legacy reads, and the test
asserts they are excluded from the palette.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/api/routes/device_ws.py

@@ -42,8 +42,9 @@ from sqlalchemy import update
 from app.api.routes.agent_setup import handle_journey_message, handle_journey_start
 from app.config.settings import settings
 from app.core.agent_runner import trigger_pending_runs
+from app.core.agent_session_buffer import session_buffer
 from app.core.brief_agent import run_home_brief, run_project_brief
-from app.core.deep_agent import run_floating_stream, run_home_stream, run_task_brief_research_stream
+from app.core.deep_agent import run_contextual_stream, run_floating_stream, run_home_stream, run_task_brief_research_stream
 from app.core.output_formatter import extract_canvas_block
 from app.core.device_manager import device_manager
 from app.core.memory_middleware import MemoryMiddleware
@@ -52,6 +53,7 @@ from app.core.ws_context import clear_client_executor, set_client_executor
 from app.db import async_session
 from app.models import AgentRunLog
 from app.schemas import WsFrameType, WsStreamEnd
+from app.schemas.contextual import ContextualScope, render_scope_block
 
 logger = logging.getLogger(__name__)
 
@@ -197,6 +199,16 @@ async def _message_loop(websocket: WebSocket, user_id: str) -> None:
         elif frame_type == WsFrameType.index_session_cancel:
             await _handle_index_session_cancel(websocket, frame)
 
+        elif frame_type == WsFrameType.contextual_request:
+            asyncio.create_task(
+                _handle_contextual_request(websocket, user_id, frame)
+            )
+
+        elif frame_type == WsFrameType.contextual_scope_update:
+            asyncio.create_task(
+                _handle_contextual_scope_update(websocket, user_id, frame)
+            )
+
         elif frame_type == "pong":
             # Heartbeat ack — nothing to do, connection is alive.
             pass
@@ -359,6 +371,122 @@ async def _handle_floating_request(
     )
 
 
+# ── v8 Contextual Sidebar Handlers ───────────────────────────────────
+
+
+def get_session_buffer(user_id: str, session_id: str, channel: str = "contextual"):
+    """Return a session-scoped buffer proxy for the given user+session.
+
+    Returns a _ContextualBufferProxy that exposes append_system_message().
+    Defined at module level so tests can monkeypatch it.
+    The channel kwarg is accepted for forward-compatibility.
+    """
+    from app.core.agent_session_buffer import ContextualBufferProxy  # noqa: PLC0415
+    return ContextualBufferProxy(session_buffer, user_id, session_id)
+
+
+async def _handle_contextual_request(
+    websocket: WebSocket,
+    user_id: str,
+    frame: dict,
+) -> None:
+    """Handle a contextual_request frame — runs the contextual agent and streams frames."""
+    request_id = frame.get("request_id") or str(uuid4())
+    message: str = frame.get("message", "")
+    session_id: str = frame.get("session_id") or str(uuid4())
+    scope_payload: dict = frame.get("scope", {})
+    logger.info(
+        "device_ws: contextual_request_start user=%s req=%s session=%s msg=%s",
+        user_id,
+        request_id,
+        session_id,
+        message[:200],
+    )
+
+    scope = ContextualScope.model_validate(scope_payload)
+
+    # Enrich context with memory before the LLM call.
+    async with async_session() as db:
+        memory = MemoryMiddleware(db)
+        memory_context = await memory.enrich_context(
+            user_id,
+            message,
+            trace_id=request_id,
+            session_id=session_id,
+        )
+
+    context: dict = {
+        "conversation_history": frame.get("conversation_history", []),
+        "format_prefs": frame.get("format_prefs"),
+        "_debug": {"request_id": request_id, "session_id": session_id, "user_id": user_id},
+        **memory_context,
+    }
+
+    executor = await _make_ws_executor(websocket, user_id)
+    set_client_executor(executor)
+    response_chunks: list[str] = []
+    try:
+        event_stream = run_contextual_stream(
+            user_id=user_id,
+            message=message,
+            context=context,
+            scope=scope,
+        )
+        formatter = StreamFormatter(request_id=request_id)
+        async for ws_frame in formatter.format(event_stream):
+            await websocket.send_text(ws_frame.model_dump_json())
+            if ws_frame.type == "stream_text":  # type: ignore[union-attr]
+                response_chunks.append(ws_frame.chunk)  # type: ignore[union-attr]
+    except Exception as exc:
+        logger.error(
+            "device_ws: contextual_request failed user=%s req=%s: %s",
+            user_id, request_id, exc,
+        )
+    finally:
+        clear_client_executor()
+
+    # Store episode so the contextual agent can recall prior turns.
+    async with async_session() as db:
+        memory = MemoryMiddleware(db)
+        await memory.store_episode(
+            user_id, session_id, message, "".join(response_chunks), trace_id=request_id
+        )
+    logger.info(
+        "device_ws: contextual_request_end user=%s req=%s session=%s response_chars=%d",
+        user_id,
+        request_id,
+        session_id,
+        len("".join(response_chunks)),
+    )
+
+
+async def _handle_contextual_scope_update(
+    websocket: WebSocket,
+    user_id: str,
+    frame: dict,
+) -> None:
+    """Handle a contextual_scope_update frame.
+
+    Injects a synthetic system message into the session buffer so the next
+    agent turn knows the user navigated.  No LLM call is made.
+    """
+    session_id: str = frame.get("session_id") or str(uuid4())
+    scope = ContextualScope.model_validate(frame.get("scope", {}))
+    block = render_scope_block(scope)
+    buf = get_session_buffer(user_id, session_id, channel="contextual")
+    buf.append_system_message(
+        f"User navigated to a new view. {block} Treat this as the new active context."
+    )
+    await websocket.send_text(json.dumps({
+        "type": WsFrameType.contextual_scope_ack,
+        "session_id": session_id,
+    }))
+    logger.info(
+        "device_ws: contextual_scope_update user=%s session=%s page=%s",
+        user_id, session_id, scope.page,
+    )
+
+
 async def _handle_brief_request(
     websocket: WebSocket,
     user_id: str,

app/core/agent_session_buffer.py

@@ -54,6 +54,43 @@ class _SessionBuffer:
         with self._lock:
             self._store.pop((user_id, session_id), None)
 
+    def append_system_message(self, user_id: str, session_id: str, text: str) -> None:
+        """Append a synthetic system message to the buffer for the given session.
+
+        Creates the session slot if it does not yet exist.  Used by the
+        contextual_scope_update handler to inject navigation events without
+        making an LLM call.
+        """
+        from langchain_core.messages import SystemMessage  # noqa: PLC0415
+
+        key = (user_id, session_id)
+        with self._lock:
+            entry = self._store.get(key)
+            if entry is None:
+                msgs: list[BaseMessage] = [SystemMessage(content=text)]
+            else:
+                _, existing = entry
+                msgs = list(existing) + [SystemMessage(content=text)]
+            capped = msgs[-MAX_MESSAGES_PER_SESSION:]
+            self._store[key] = (time.monotonic(), capped)
+
+
+class ContextualBufferProxy:
+    """Thin wrapper around _SessionBuffer that closes over user_id + session_id.
+
+    Returned by get_session_buffer() so callers can call
+    ``proxy.append_system_message(text)`` without threading user_id/session_id
+    through every call site.
+    """
+
+    def __init__(self, buf: "_SessionBuffer", user_id: str, session_id: str) -> None:
+        self._buf = buf
+        self._user_id = user_id
+        self._session_id = session_id
+
+    def append_system_message(self, text: str) -> None:
+        self._buf.append_system_message(self._user_id, self._session_id, text)
+
 
 # Module-level singleton — same pattern as _pending_states in api/app/api/routes/auth.py
 session_buffer = _SessionBuffer()

app/core/deep_agent.py

@@ -392,6 +392,26 @@ For specific dates not listed, compute local-midnight in the user timezone and c
 {request_context}\
 """
 
+_CONTEXTUAL_SYSTEM_PROMPT = """You are adiuvAI's contextual assistant. The user is working inside the app and has opened a side chat anchored to a specific view ("current view"). Help them act on that view: recap, plan, create entities, answer questions.
+
+Rules:
+1. Base context (current view summary) is provided every turn. Treat it as ground truth for ids and names; never invent them.
+2. ALL reads go through `get_page_details`. The legacy tools `list_projects`, `get_project`, `list_tasks`, `get_task`, `list_notes`, `get_note` are NOT available in this channel — do not attempt to call them. To find an entity by name, call `get_page_details({entityType: 'projects_all' | 'tasks_all' | 'timeline_all'})` to list, then `get_page_details({entityType: '<type>', entityId})` for the full snapshot.
+3. When the user requests an action that creates or updates an entity:
+   - If the current view is a project and no project is specified, use the current project automatically.
+   - If the current view is the global Tasks / Projects / Timeline list and no project is specified, ASK before attaching to any project. Don't silently create orphan entities.
+4. The current view can change mid-conversation (user navigates). When you see a system message "User navigated to ...", treat the new view as the active context. Prior turns remain visible but the active scope shifts.
+5. Notes: you can read note bodies via `get_page_details({entityType:'note'})`. You CANNOT edit, summarize-to-replace, or append. Tell the user "note editing is coming in a later release" if asked.
+6. Be concise. Default to 1-3 short paragraphs. Bullet lists fine. Don't restate the user's request.
+7. Never expose ids in prose. Use names. Ids only travel through tool calls.
+
+# Date context
+{date_context}
+
+# Language
+{language_instruction}
+"""
+
 _TASK_BRIEF_RESEARCH_SYSTEM_PROMPT = """\
 You are an executive assistant preparing a briefing dossier for your principal before they act on a specific task.
 Your job: gather all relevant context, synthesize it into a tight actionable dossier, and — if the task requires writing (email, message, document) — produce a ready-to-use draft.{user_identity}
@@ -556,6 +576,55 @@ def _all_tools() -> list[Any]:
     return [*TASK_TOOLS, *PROJECT_TOOLS, *NOTE_TOOLS, *TIMELINE_TOOLS]
 
 
+# ── Contextual sidebar tools ──────────────────────────────────────────
+
+
+@tool
+async def get_page_details(
+    entity_type: str = "",
+    entity_id: str = "",
+) -> str:
+    """Fetch full details for the entity currently in view.
+
+    entity_type: one of 'project' | 'task' | 'note' | 'timeline_event' |
+                 'tasks_all' | 'projects_all' | 'timeline_all'.
+    entity_id: UUID of the entity for singular entity views.  Omit for list views.
+
+    The Electron drizzle-executor fulfils this op against local SQLite and
+    returns the row(s) as a JSON tool result.
+    """
+    result = await execute_on_client(
+        action="get_page_details",
+        table=entity_type or "unknown",
+        data={"entityId": entity_id or None},
+    )
+    if not result:
+        return "No details found."
+    return str(result)
+
+
+def _contextual_tools(user_id: str, trace_id: str | None) -> list[Any]:
+    """Return the tool palette for the contextual sidebar agent.
+
+    Read ops go through get_page_details only — legacy list_*/get_* tools
+    return shallow snapshots and cause the agent to under-answer (see
+    smoke trace 0b46841484ba7d024ed9f8d5ac8b1df0). Writes are limited
+    to entity creation + task update; note edits are next-sprint.
+    """
+    from app.agents.note_agent import create_note  # noqa: PLC0415
+    from app.agents.task_agent import create_task, update_task  # noqa: PLC0415
+    from app.agents.timeline_agent import create_timeline  # noqa: PLC0415
+
+    return [
+        get_page_details,
+        create_task,
+        update_task,
+        create_note,
+        create_timeline,
+        *_memory_tools(user_id, trace_id),
+    ]
+
+
 def _trace_id_from_context(context: dict[str, Any]) -> str | None:
     debug = context.get("_debug")
     if isinstance(debug, dict):
@@ -1522,6 +1591,49 @@ async def run_floating_stream(
         yield "token", _fallback_from_raw_floating_text("".join(raw_chunks))
 
 
+async def run_contextual_stream(
+    user_id: str,
+    message: str,
+    context: dict[str, Any],
+    scope: "ContextualScope",  # type: ignore[name-defined]
+) -> AsyncGenerator[tuple[str, Any], None]:
+    """Run the contextual agent for a single user turn.
+
+    Mirrors run_floating_stream's plumbing but injects the rendered scope
+    block into the system prompt and exposes the contextual tool set.
+    Note-edit tools (propose_note_edit) are intentionally excluded.
+
+    *context contract*: callers MUST include ``context["_debug"]["session_id"]``
+    (a non-empty str) so that ``_session_id_from_context`` can extract it for
+    tracing and episode storage downstream.  The WS handler in device_ws.py
+    satisfies this by always populating ``_debug`` before calling this function.
+    """
+    from app.schemas.contextual import ContextualScope, render_scope_block  # noqa: PLC0415
+
+    prepared_context = await _prepare_context(message, context)
+    trace_id = _trace_id_from_context(prepared_context)
+
+    system_prompt, langfuse_prompt = _build_system_prompt(
+        "contextual_system", _CONTEXTUAL_SYSTEM_PROMPT, prepared_context,
+    )
+    scope_block = render_scope_block(scope)
+    system_prompt = system_prompt + f"\n\n## Current view\n{scope_block}"
+
+    tools = _contextual_tools(user_id, trace_id)
+
+    async for event in _run_single_agent_stream(
+        user_id=user_id,
+        system_prompt=system_prompt,
+        message=message,
+        context=prepared_context,
+        langfuse_prompt=langfuse_prompt,
+        agent_name="contextual-agent",
+        tools=tools,
+        conversation_history=context.get("conversation_history"),
+    ):
+        yield event
+
+
 async def run_task_brief_research_stream(
     user_id: str,
     task_id: str,

app/schemas/__init__.py

@@ -96,6 +96,10 @@ class WsFrameType(str, Enum):
     index_file_result = "index_file_result"
     index_session_progress = "index_session_progress"
     index_session_done = "index_session_done"
+    # ── v8 contextual sidebar frame types ────────────────────────────
+    contextual_request = "contextual_request"
+    contextual_scope_update = "contextual_scope_update"
+    contextual_scope_ack = "contextual_scope_ack"
 
 
 class WsToolCall(BaseModel):

app/schemas/contextual.py

@@ -0,0 +1,73 @@
+"""Contextual sidebar scope schema and prompt block renderer.
+
+ContextualScope mirrors the TypeScript ContextualScope type sent by the
+Electron renderer when the user opens the side chat anchored to a specific
+view.  The renderer ships camelCase keys; Pydantic's alias_generator maps
+them to snake_case Python attributes automatically.
+"""
+
+from __future__ import annotations
+
+from typing import Literal, Optional
+
+from pydantic import BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+
+
+PageType = Literal[
+    "timeline",
+    "tasks",
+    "projects-list",
+    "project",
+    "note",
+]
+
+EntityType = Literal["project", "note", "task", "timeline_event"]
+
+
+class ContextualScope(BaseModel):
+    """Scope payload sent by the Electron renderer for contextual chat.
+
+    The renderer ships camelCase keys (entityType, entityId, ...).  Pydantic's
+    alias generator maps them to snake_case Python attrs.
+    """
+
+    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
+
+    page: PageType
+    entity_type: Optional[EntityType] = None
+    entity_id: Optional[str] = None
+    entity_name: Optional[str] = None
+    project_id: Optional[str] = None
+    char_count: Optional[int] = None
+    counts: Optional[dict[str, int]] = None
+    filters: Optional[dict] = None
+
+
+def render_scope_block(scope: ContextualScope) -> str:
+    """Produce a single-paragraph human-readable summary of the current view
+    for injection into the contextual agent system prompt.
+
+    Never emits internal ids — only names.  The LLM is told to use names in
+    prose; ids travel through tool calls.
+    """
+    if scope.entity_type == "project":
+        c = scope.counts or {}
+        return (
+            f"User is viewing the project {scope.entity_name!r}. "
+            f"{c.get('tasks', 0)} tasks, "
+            f"{c.get('notes', 0)} notes, "
+            f"{c.get('milestones', 0)} milestones."
+        )
+    if scope.entity_type == "note":
+        return (
+            f"User is viewing the note {scope.entity_name!r} "
+            f"({scope.char_count or 0} characters)."
+        )
+    if scope.page == "tasks":
+        return "User is viewing the global Tasks list (all projects)."
+    if scope.page == "timeline":
+        return "User is viewing the global Timeline view."
+    if scope.page == "projects-list":
+        return "User is viewing the Projects list."
+    return f"User is on page {scope.page}."

tests/test_contextual_scope.py

@@ -0,0 +1,52 @@
+import pytest
+from app.schemas.contextual import ContextualScope, render_scope_block
+
+
+def test_render_project_scope():
+    scope = ContextualScope(
+        page="project",
+        entity_type="project",
+        entity_id="p1",
+        entity_name="Acme Q3 launch",
+        counts={"tasks": 12, "notes": 4, "milestones": 3},
+    )
+    block = render_scope_block(scope)
+    assert "Acme Q3 launch" in block
+    assert "12 tasks" in block
+    assert "4 notes" in block
+    assert "3 milestones" in block
+    assert "p1" not in block
+
+
+def test_render_list_scope_no_entity():
+    scope = ContextualScope(page="tasks", entity_type=None)
+    block = render_scope_block(scope)
+    assert "tasks" in block.lower()
+    assert "None" not in block
+
+
+def test_render_note_scope_includes_char_count():
+    scope = ContextualScope(
+        page="note",
+        entity_type="note",
+        entity_id="n1",
+        entity_name="Meeting 14 May",
+        project_id="p1",
+        char_count=4280,
+    )
+    block = render_scope_block(scope)
+    assert "Meeting 14 May" in block
+    assert "4280" in block or "4,280" in block
+
+
+def test_parses_camelcase_payload_from_renderer():
+    payload = {
+        "page": "project",
+        "entityType": "project",
+        "entityId": "p1",
+        "entityName": "Acme",
+        "counts": {"tasks": 5, "notes": 1, "milestones": 2},
+    }
+    scope = ContextualScope.model_validate(payload)
+    assert scope.entity_id == "p1"
+    assert scope.entity_name == "Acme"

tests/test_contextual_ws.py

@@ -0,0 +1,44 @@
+"""Tests for contextual WS frame handlers.
+
+These tests only exercise the new handler functions in device_ws.py and do
+not depend on litellm or the full deep_agent import chain.  They monkeypatch
+run_contextual_stream so no LLM call is made.
+"""
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+@pytest.mark.asyncio
+async def test_handle_contextual_scope_update_appends_system_message_no_llm(monkeypatch):
+    """_handle_contextual_scope_update must:
+    - call append_system_message on the session buffer
+    - send a contextual_scope_ack back on the socket
+    - make no LLM call
+    """
+    from app.api.routes import device_ws
+
+    ws = AsyncMock()
+    buffer = MagicMock()
+    buffer.append_system_message = MagicMock()
+
+    payload = {
+        "type": "contextual_scope_update",
+        "session_id": "s1",
+        "scope": {
+            "page": "project",
+            "entityType": "project",
+            "entityId": "p1",
+            "entityName": "Acme",
+            "counts": {"tasks": 1, "notes": 0, "milestones": 0},
+        },
+    }
+
+    monkeypatch.setattr(device_ws, "get_session_buffer", lambda *a, **kw: buffer)
+    await device_ws._handle_contextual_scope_update(ws, "user1", payload)
+
+    ws.send_text.assert_awaited_once()
+    import json
+    sent = json.loads(ws.send_text.await_args.args[0])
+    assert sent["type"] == "contextual_scope_ack"
+    assert sent["session_id"] == "s1"
+    buffer.append_system_message.assert_called_once()

tests/test_run_contextual.py

@@ -0,0 +1,85 @@
+"""Tests for run_contextual_stream.
+
+These tests monkeypatch _run_single_agent_stream (the actual internal runner)
+rather than the plan's fictional _run_agent_loop, matching the real
+deep_agent.py architecture.
+"""
+import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
+from app.schemas.contextual import ContextualScope
+
+
+@pytest.mark.asyncio
+async def test_run_contextual_stream_includes_scope_block(monkeypatch):
+    """run_contextual_stream must inject the scope block into the system prompt
+    and include get_page_details in the tool list while excluding note-edit tools."""
+    import app.core.deep_agent as deep_agent
+
+    captured = {}
+
+    async def fake_stream(
+        *,
+        user_id,
+        system_prompt,
+        message,
+        context,
+        agent_name="agent",
+        tools=None,
+        conversation_history=None,
+        **kwargs,
+    ):
+        captured["sys"] = system_prompt
+        captured["tool_names"] = [getattr(t, "name", str(t)) for t in (tools or [])]
+        captured["agent_name"] = agent_name
+        # Async generator that yields nothing — still satisfies the protocol.
+        if False:
+            yield  # pragma: no cover
+
+    monkeypatch.setattr(deep_agent, "_run_single_agent_stream", fake_stream)
+
+    scope = ContextualScope(
+        page="project",
+        entity_type="project",
+        entity_id="p1",
+        entity_name="Acme",
+        counts={"tasks": 1, "notes": 0, "milestones": 0},
+    )
+
+    context = {
+        "conversation_history": [],
+        "_debug": {"session_id": "s1"},
+    }
+
+    results = []
+    async for item in deep_agent.run_contextual_stream(
+        user_id="user1",
+        message="hi",
+        context=context,
+        scope=scope,
+    ):
+        results.append(item)
+
+    assert "Acme" in captured["sys"], "scope block must appear in system prompt"
+    assert "Current view" in captured["sys"], "section header must be present"
+
+    names = captured["tool_names"]
+    assert "get_page_details" in names, "get_page_details tool must be included"
+
+    # Entity-create tools: at least one of these must be present.
+    assert any(n in names for n in ("create_task", "create_note", "update_task")), (
+        "at least one entity-create tool must be present"
+    )
+
+    assert "create_timeline" in names, "create_timeline tool must be included"
+
+    # Note edit tools must NOT be exposed.
+    assert "propose_note_edit" not in names, "propose_note_edit must be excluded"
+
+    # Legacy read tools must be excluded — they return shallow snapshots and
+    # cause the agent to under-answer (see trace 0b46841484ba7d024ed9f8d5ac8b1df0).
+    assert "list_projects" not in names, "list_projects must be excluded (legacy read)"
+    assert "get_project" not in names, "get_project must be excluded (legacy read)"
+    assert "list_tasks" not in names, "list_tasks must be excluded (legacy read)"
+    assert "get_task" not in names, "get_task must be excluded (legacy read)"
+    assert "list_notes" not in names, "list_notes must be excluded (legacy read)"
+    assert "get_note" not in names, "get_note must be excluded (legacy read)"