feat(server): integrate MCP queries and investigation messaging

Add project-bound OAuth MCP queries and controlled analysis with scoped evidence,
limits, timeouts and a nearby-model spatial query using EPSG:4326 inputs.
Default nearby searches to 200 meters and five results.

Integrate investigation messaging/presentation, health/readiness and typed history;
remove duplicate web search and fix online simulation and compressed baseline writes.
Include project migrations 023-025; keep local credentials and demo seeding excluded.

Validation: 319 focused pytest checks; real PostGIS temporary-table regression;
read-only Jinqian bridge radius comparison; cross-language MCP discovery/calls.

app/infra/db/postgresql/agent_messages.py

@@ -0,0 +1,88 @@
+"""Project scoped private message storage, sharing task transaction locks."""
+from psycopg import AsyncConnection
+
+from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository
+
+
+class AgentMessageRepository(OperationalTaskRepository):
+    def __init__(self, connection: AsyncConnection, project_id):
+        super().__init__(connection, project_id)
+
+    async def get_message(self, message_id: str) -> dict | None:
+        cursor = await self.connection.execute(
+            "SELECT * FROM workbench.agent_messages WHERE project_id=%s AND message_id=%s",
+            (self.project_id, message_id),
+        )
+        row = await cursor.fetchone()
+        return self._message(row) if row else None
+
+    async def get_message_request(self, owner_id: str, session_id: str, request_id: str) -> dict | None:
+        cursor = await self.connection.execute(
+            """SELECT * FROM workbench.agent_messages WHERE project_id=%s
+               AND source_owner_id=%s AND source_session_id=%s AND request_id=%s""",
+            (self.project_id, owner_id, session_id, request_id),
+        )
+        row = await cursor.fetchone()
+        return self._message(row) if row else None
+
+    @staticmethod
+    def _message(row) -> dict:
+        return dict(row)
+
+    async def insert_message(self, message: dict) -> None:
+        keys = [key for key in message if key != "project_id"]
+        await self.connection.execute(
+            f"INSERT INTO workbench.agent_messages (project_id,{','.join(keys)}) "
+            f"VALUES ({','.join(['%s'] * (len(keys) + 1))})",
+            (self.project_id, *(message[key] for key in keys)),
+        )
+
+    async def list_session_messages(
+        self, owner_id: str, session_id: str, limit: int, *, cursor_at=None, cursor_id=None, pending_only=False,
+    ) -> list[dict]:
+        cursor = await self.connection.execute(
+            """SELECT * FROM workbench.agent_messages WHERE project_id=%s
+               AND conversation_owner_id=%s AND (source_session_id=%s OR
+                    (direction='to_chat' AND target_session_id=%s))
+               AND (%s::timestamptz IS NULL OR (created_at,message_id)<(%s::timestamptz,%s::uuid))
+               AND (NOT %s OR status IN ('pending','enqueued'))
+               ORDER BY created_at DESC,message_id DESC LIMIT %s""",
+            (self.project_id, owner_id, session_id, session_id, cursor_at, cursor_at, cursor_id, pending_only, limit),
+        )
+        return [self._message(row) for row in await cursor.fetchall()]
+
+    async def list_run_session_messages(
+        self, session_id: str, limit: int, *, cursor_at=None, cursor_id=None, pending_only=False,
+    ) -> list[dict]:
+        cursor = await self.connection.execute(
+            """SELECT m.* FROM workbench.agent_messages m
+               JOIN workbench.operational_task_runs r ON r.project_id=m.project_id AND r.run_id=m.run_id
+               WHERE m.project_id=%s AND r.agent_session_id=%s
+                 AND r.status IN ('claimed','running')
+                 AND (%s::timestamptz IS NULL OR (m.created_at,m.message_id)<(%s::timestamptz,%s::uuid))
+                 AND (NOT %s OR (m.direction='to_task' AND m.status IN ('pending','enqueued')))
+               ORDER BY m.created_at DESC,m.message_id DESC LIMIT %s""",
+            (self.project_id, session_id, cursor_at, cursor_at, cursor_id, pending_only, limit),
+        )
+        return [self._message(row) for row in await cursor.fetchall()]
+
+    async def list_pending_messages(self, limit: int, *, cursor_at=None, cursor_id=None) -> list[dict]:
+        cursor = await self.connection.execute(
+            """SELECT m.* FROM workbench.agent_messages m
+               JOIN workbench.operational_task_runs r
+                 ON r.project_id=m.project_id AND r.run_id=m.run_id
+               WHERE m.project_id=%s AND m.status IN ('pending','enqueued')
+                 AND (m.direction='to_chat' OR r.status IN ('claimed','running'))
+                 AND (%s::timestamptz IS NULL OR (m.created_at,m.message_id)>(%s::timestamptz,%s::uuid))
+               ORDER BY m.created_at,m.message_id LIMIT %s""",
+            (self.project_id, cursor_at, cursor_at, cursor_id, limit),
+        )
+        return [self._message(row) for row in await cursor.fetchall()]
+
+    async def update_message_delivery(self, message_id: str, update: dict) -> None:
+        keys = list(update)
+        await self.connection.execute(
+            f"UPDATE workbench.agent_messages SET {','.join(key+'=%s' for key in keys)} "
+            "WHERE project_id=%s AND message_id=%s",
+            (*(update[key] for key in keys), self.project_id, message_id),
+        )

app/infra/db/postgresql/analysis.py

@@ -221,6 +221,7 @@ class AnalysisRepository:
         *,
         limit: int = 100,
         offset: int = 0,
+        lookahead: bool = False,
     ) -> list[dict]:
         if not 1 <= limit <= 500 or offset < 0:
             raise ValueError("analysis result page is out of range")
@@ -235,7 +236,7 @@ class AnalysisRepository:
             query += " AND result_type = %s"
             params = (run_id, result_type)
         query += " ORDER BY created_at, result_id LIMIT %s OFFSET %s"
-        params += (limit, offset)
+        params += (limit + int(lookahead), offset)
 
         async with conn.cursor() as cur:
             await cur.execute(query, params)

app/infra/db/postgresql/mcp_queries.py

@@ -0,0 +1,91 @@
+"""Persistent MCP context and receipt storage in the selected project's BizDB."""
+
+from datetime import datetime, timezone
+from psycopg import AsyncConnection
+from psycopg.types.json import Jsonb
+from app.mcp.models import BoundContext
+
+
+class McpQueryRepository:
+    def __init__(self, conn: AsyncConnection, project_id):
+        self.conn, self.project_id = conn, project_id
+
+    async def create(self, context: BoundContext):
+        async with self.conn.transaction(), self.conn.cursor() as cur:
+            # Bounded maintenance; expired/revoked handles remain invalid even if cleanup fails.
+            await cur.execute(
+                """DELETE FROM workbench.mcp_query_contexts WHERE context_handle IN
+                (SELECT context_handle FROM workbench.mcp_query_contexts
+                 WHERE project_id=%s AND (expires_at < now() OR revoked_at IS NOT NULL)
+                 ORDER BY expires_at LIMIT 100 FOR UPDATE SKIP LOCKED)""",
+                (self.project_id,),
+            )
+            await cur.execute(
+                """INSERT INTO workbench.mcp_query_contexts
+                (context_handle,project_id,subject,client_id,document,expires_at)
+                VALUES (%s,%s,%s,%s,%s,%s)""",
+                (
+                    context.context_handle,
+                    self.project_id,
+                    context.subject,
+                    context.client_id,
+                    Jsonb(context.model_dump(mode="json")),
+                    context.expires_at,
+                ),
+            )
+
+    async def get(self, handle: str) -> BoundContext | None:
+        async with self.conn.transaction(), self.conn.cursor() as cur:
+            await cur.execute(
+                """SELECT document,revoked_at FROM workbench.mcp_query_contexts
+                WHERE context_handle=%s AND project_id=%s""",
+                (handle, self.project_id),
+            )
+            row = await cur.fetchone()
+        if not row:
+            return None
+        document = dict(row["document"])
+        document["revoked_at"] = row["revoked_at"]
+        return BoundContext.model_validate(document)
+
+    async def revoke(self, handle: str, subject: str, client_id: str):
+        async with self.conn.transaction(), self.conn.cursor() as cur:
+            await cur.execute(
+                """UPDATE workbench.mcp_query_contexts SET revoked_at=COALESCE(revoked_at,%s)
+                WHERE context_handle=%s AND project_id=%s AND subject=%s AND client_id=%s""",
+                (
+                    datetime.now(timezone.utc),
+                    handle,
+                    self.project_id,
+                    subject,
+                    client_id,
+                ),
+            )
+
+    async def receipt(self, scope_key: str, call_id: str) -> dict | None:
+        async with self.conn.transaction(), self.conn.cursor() as cur:
+            await cur.execute(
+                """SELECT parameters,result FROM workbench.mcp_query_receipts
+                WHERE project_id=%s AND scope_key=%s AND call_id=%s""",
+                (self.project_id, scope_key, call_id),
+            )
+            return await cur.fetchone()
+
+    async def save_receipt(
+        self, scope_key: str, call_id: str, parameters: dict, result: dict
+    ):
+        # Caller owns the short evidence + receipt transaction.
+        async with self.conn.cursor() as cur:
+            await cur.execute(
+                """INSERT INTO workbench.mcp_query_receipts
+                (project_id,scope_key,call_id,parameters,result) VALUES (%s,%s,%s,%s,%s)
+                ON CONFLICT (project_id,scope_key,call_id) DO NOTHING""",
+                (self.project_id, scope_key, call_id, Jsonb(parameters), Jsonb(result)),
+            )
+
+    async def lock_receipt(self, scope_key: str, call_id: str):
+        async with self.conn.cursor() as cur:
+            await cur.execute(
+                "SELECT pg_advisory_xact_lock(hashtextextended(%s,0))",
+                (f"mcp:{self.project_id}:{scope_key}:{call_id}",),
+            )

app/infra/db/postgresql/operational_tasks.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from contextlib import asynccontextmanager
+from datetime import datetime
 from typing import Any
 
 from psycopg import AsyncConnection
@@ -11,6 +12,7 @@ from uuid import uuid4
 
 from app.domain.domain_errors import DomainError
 from app.infra.db.postgresql.workbench_events import lock_event_transaction
+from app.infra.db.postgresql.workbench_events import WorkbenchEventRepository
 
 
 class OperationalTaskRepository:
@@ -37,6 +39,13 @@ class OperationalTaskRepository:
             (f"tjwater-operational:{self.project_id}:{key}",),
         )
 
+    async def try_lock(self, key: str) -> bool:
+        cursor = await self.connection.execute(
+            "SELECT pg_try_advisory_xact_lock(hashtextextended(%s, 0)) AS acquired",
+            (f"tjwater-operational:{self.project_id}:{key}",),
+        )
+        return bool((await cursor.fetchone())["acquired"])
+
     async def get_request(self, scope: str, request_id: str) -> dict | None:
         cursor = await self.connection.execute(
             """SELECT request_digest,response
@@ -103,6 +112,50 @@ class OperationalTaskRepository:
         row = await cursor.fetchone()
         return dict(row) if row else None
 
+    async def list_task_runs(self, task_id: str, limit: int, offset: int) -> list[dict]:
+        cursor = await self.connection.execute(
+            """SELECT * FROM workbench.operational_task_runs
+               WHERE project_id=%s AND task_id=%s
+               ORDER BY generation DESC LIMIT %s OFFSET %s""",
+            (self.project_id, task_id, limit, offset),
+        )
+        return [{key: value for key, value in row.items() if key != "project_id"}
+                for row in await cursor.fetchall()]
+
+    async def get_latest_run(self, task_id: str) -> dict | None:
+        rows = await self.list_task_runs(task_id, 1, 0)
+        return rows[0] if rows else None
+
+    async def require_execution_storage(self) -> None:
+        """Reject claims before changing state if the inbox cannot be used."""
+        cursor = await self.connection.execute(
+            "SELECT to_regclass('workbench.agent_messages') IS NOT NULL AS available"
+        )
+        if not (await cursor.fetchone())["available"]:
+            raise DomainError(
+                "DEPENDENCY_UNAVAILABLE",
+                "Agent message storage is not provisioned; apply "
+                "resources/sql/024_agent_messages.sql to this project database before starting runs",
+                status=503,
+            )
+
+    async def pending_run_messages(self, run_id: str) -> list[str]:
+        cursor = await self.connection.execute(
+            """SELECT message_id FROM workbench.agent_messages
+               WHERE project_id=%s AND run_id=%s AND direction='to_task'
+                 AND status IN ('pending','enqueued') ORDER BY created_at LIMIT 100""",
+            (self.project_id, run_id),
+        )
+        return [str(row["message_id"]) for row in await cursor.fetchall()]
+
+    async def fail_run_messages(self, run_id: str, failure_detail: str, now: datetime) -> None:
+        await self.connection.execute(
+            """UPDATE workbench.agent_messages SET status='failed',failure_detail=%s,updated_at=%s
+               WHERE project_id=%s AND run_id=%s AND direction='to_task'
+                 AND status IN ('pending','enqueued')""",
+            (failure_detail, now, self.project_id, run_id),
+        )
+
     async def insert_run(self, run: dict) -> None:
         await self.connection.execute(
             """INSERT INTO workbench.operational_task_runs
@@ -142,7 +195,8 @@ class OperationalTaskRepository:
             raise DomainError("VERSION_CONFLICT", "Run revision changed", status=409)
 
     async def list_runs(
-        self, *, status: str | None, limit: int, cursor_created_at=None, cursor_run_id=None
+        self, *, status: str | None, limit: int, cursor_created_at=None, cursor_run_id=None,
+        task_type: str | None = None, task_created_from: datetime | None = None,
     ) -> list[dict]:
         result = await self.connection.execute(
             """SELECT run_id,task_id,generation,execution_request_id,status,revision,
@@ -154,15 +208,25 @@ class OperationalTaskRepository:
                  AND (%s::text IS NULL OR status=%s::text)
                  AND (%s::timestamptz IS NULL OR (created_at,run_id)>(%s::timestamptz,%s::uuid))
                  AND (status<>'queued' OR next_dispatch_at<=now())
+                 AND EXISTS (
+                     SELECT 1 FROM workbench.operational_tasks t
+                     WHERE t.project_id=operational_task_runs.project_id
+                       AND t.task_id=operational_task_runs.task_id
+                       AND (%s::text IS NULL OR t.task_type=%s::text)
+                       AND (%s::timestamptz IS NULL OR t.created_at>=%s::timestamptz)
+                 )
                ORDER BY created_at,run_id LIMIT %s""",
             (
                 self.project_id, status, status, cursor_created_at,
-                cursor_created_at, cursor_run_id, limit,
+                cursor_created_at, cursor_run_id,
+                task_type, task_type, task_created_from, task_created_from, limit,
             ),
         )
         return [dict(row) for row in await result.fetchall()]
 
-    async def claim_expired_runs(self, limit: int) -> list[dict]:
+    async def claim_expired_runs(
+        self, limit: int, *, task_type: str | None = None, task_created_from: datetime | None = None,
+    ) -> list[dict]:
         cursor = await self.connection.execute(
             """SELECT run_id,task_id,generation,execution_request_id,status,revision,
                       agent_session_id,lease_expires_at,heartbeat_at,started_at,finished_at,
@@ -171,8 +235,15 @@ class OperationalTaskRepository:
                FROM workbench.operational_task_runs
                WHERE project_id=%s AND status IN ('claimed','running')
                  AND lease_expires_at IS NOT NULL AND lease_expires_at<=now()
+                 AND EXISTS (
+                     SELECT 1 FROM workbench.operational_tasks t
+                     WHERE t.project_id=operational_task_runs.project_id
+                       AND t.task_id=operational_task_runs.task_id
+                       AND (%s::text IS NULL OR t.task_type=%s::text)
+                       AND (%s::timestamptz IS NULL OR t.created_at>=%s::timestamptz)
+                 )
                ORDER BY lease_expires_at,run_id FOR UPDATE SKIP LOCKED LIMIT %s""",
-            (self.project_id, limit),
+            (self.project_id, task_type, task_type, task_created_from, task_created_from, limit),
         )
         return [dict(row) for row in await cursor.fetchall()]
 
@@ -212,6 +283,9 @@ class OperationalTaskRepository:
         )
         return event_seq
 
+    async def current_event_cursor(self) -> int:
+        return await WorkbenchEventRepository(self.connection, self.project_id).current_cursor()
+
     async def get_task_by_create_request(self, request_id: str) -> dict | None:
         cursor = await self.connection.execute(
             """SELECT document FROM workbench.operational_tasks
@@ -246,7 +320,9 @@ class OperationalTaskRepository:
         cursor = await self.connection.execute(
             """SELECT task.document,
                       CASE WHEN active_run.run_id IS NULL THEN NULL
-                           ELSE to_jsonb(active_run) - 'project_id' END AS active_run
+                           ELSE to_jsonb(active_run) - 'project_id' END AS active_run,
+                      CASE WHEN latest_run.run_id IS NULL THEN NULL
+                           ELSE to_jsonb(latest_run) - 'project_id' END AS latest_run
                FROM workbench.operational_tasks AS task
                LEFT JOIN LATERAL (
                    SELECT run.* FROM workbench.operational_task_runs AS run
@@ -254,6 +330,11 @@ class OperationalTaskRepository:
                      AND run.status IN ('queued','claimed','running')
                    ORDER BY run.generation DESC LIMIT 1
                ) AS active_run ON true
+               LEFT JOIN LATERAL (
+                   SELECT run.* FROM workbench.operational_task_runs AS run
+                   WHERE run.project_id=task.project_id AND run.task_id=task.task_id
+                   ORDER BY run.generation DESC LIMIT 1
+               ) AS latest_run ON true
                WHERE task.project_id=%s
                  AND (task.document#>>'{source,deleted_at}') IS NULL
                  AND (%s::text IS NULL OR task.task_type=%s::text)
@@ -285,7 +366,7 @@ class OperationalTaskRepository:
             ),
         )
         return [
-            {**row["document"], "active_run": row["active_run"]}
+            {**row["document"], "active_run": row["active_run"], "latest_run": row["latest_run"]}
             for row in await cursor.fetchall()
         ]
 
@@ -444,6 +525,38 @@ class OperationalTaskRepository:
     async def insert_report(self, task_id: str, snapshot: dict) -> None:
         await self._insert_artifact("operational_task_reports", task_id, snapshot)
 
+    async def get_latest_progress(self, task_id: str, generation: int) -> dict | None:
+        # Read compatibility before the coordinated progress migration is applied.
+        availability = await self.connection.execute(
+            "SELECT to_regclass('workbench.operational_task_progress') AS relation"
+        )
+        if not (await availability.fetchone())["relation"]:
+            return None
+        cursor = await self.connection.execute(
+            """SELECT progress.document FROM workbench.operational_task_progress AS progress
+               JOIN workbench.operational_task_runs AS run
+                 ON run.project_id=progress.project_id AND run.run_id=progress.run_id
+                    AND run.task_id=progress.task_id
+               WHERE progress.project_id=%s AND progress.task_id=%s AND run.generation=%s""",
+            (self.project_id, task_id, generation),
+        )
+        row = await cursor.fetchone()
+        return row["document"] if row else None
+
+    async def save_progress(self, task_id: str, snapshot: dict) -> None:
+        cursor = await self.connection.execute(
+            """INSERT INTO workbench.operational_task_progress
+               (project_id,task_id,run_id,revision,document,published_at)
+               VALUES (%s,%s,%s,%s,%s,%s)
+               ON CONFLICT(project_id,task_id,run_id) DO UPDATE SET
+                 revision=excluded.revision,document=excluded.document,published_at=excluded.published_at
+               WHERE workbench.operational_task_progress.revision=excluded.revision-1""",
+            (self.project_id, task_id, snapshot["run_id"], snapshot["revision"],
+             Jsonb(snapshot), snapshot["published_at"]),
+        )
+        if cursor.rowcount != 1:
+            raise DomainError("VERSION_CONFLICT", "Investigation progress revision changed", status=409)
+
     async def _insert_artifact(self, table: str, task_id: str, snapshot: dict) -> None:
         if table not in {"operational_task_plans", "operational_task_reports"}:
             raise ValueError("unsupported artifact table")

app/infra/db/postgresql/read_queries.py

@@ -0,0 +1,406 @@
+"""Bounded read primitives over existing project views and time-series tables."""
+
+from datetime import datetime
+from typing import Any
+from uuid import UUID
+
+from psycopg import AsyncConnection
+from psycopg import sql
+from psycopg.rows import dict_row
+from fastapi import HTTPException
+from app.infra.db.postgresql.scada import _SCADA_VIEW_SELECT, _device
+from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
+from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository
+
+ELEMENT_VIEWS = {
+    v: v + "s" for v in ("junction", "pipe", "reservoir", "tank", "pump", "valve")
+}
+MAX_QUERY_BYTES = 8 * 1024 * 1024
+
+
+async def read_rows(conn, statement, params=(), *, timeout_ms=30000):
+    async with conn.transaction(), conn.cursor(row_factory=dict_row) as cur:
+        await cur.execute(
+            "SELECT set_config('statement_timeout',%s,true)", (str(timeout_ms),)
+        )
+        await cur.execute(statement, params)
+        return await cur.fetchall()
+
+
+async def asset_rows(conn, element_type, *, element_id=None, limit=100, offset=0):
+    view = ELEMENT_VIEWS[element_type]
+    where = sql.SQL("WHERE v.id=%s") if element_id else sql.SQL("")
+    params = (element_id,) if element_id else ()
+    statement = sql.SQL(
+        "SELECT to_jsonb(v)-'geom' AS record FROM gis.{} v {} ORDER BY v.id LIMIT %s OFFSET %s"
+    ).format(sql.Identifier(view), where)
+    rows = await read_rows(conn, statement, params + (limit, offset))
+    return [row["record"] for row in rows]
+
+
+async def scada_devices(conn, *, device_id=None, limit=100, offset=0):
+    where = " WHERE id=%s" if device_id else ""
+    params = (device_id,) if device_id else ()
+    rows = await read_rows(
+        conn,
+        _SCADA_VIEW_SELECT + where + " ORDER BY device_id LIMIT %s OFFSET %s",
+        params + (limit, offset),
+    )
+    return [_device(row) for row in rows]
+
+
+async def series_rows(
+    conn,
+    *,
+    source,
+    element_type,
+    ids,
+    start=None,
+    end=None,
+    time=None,
+    field=None,
+    run_id=None,
+    limit=1000,
+    offset=0,
+    end_exclusive=False,
+):
+    node = element_type == "junction"
+    if source == "scada":
+        table, id_column, fields = (
+            "measurements",
+            "device_id",
+            {"monitored_value", "cleaned_value"},
+        )
+    else:
+        table, id_column = (
+            ("node_results", "node_id") if node else ("link_results", "link_id")
+        )
+        fields = (
+            (
+                AnalysisResultsRepository.NODE_FIELDS
+                if node
+                else AnalysisResultsRepository.LINK_FIELDS
+            )
+            if source == "analysis"
+            else (
+                RealtimeRepository.NODE_FIELDS
+                if node
+                else RealtimeRepository.LINK_FIELDS
+            )
+        )
+    if field is not None and field not in fields:
+        raise ValueError("unsupported field for this data source and element type")
+    columns = (
+        sql.SQL("{},time,{} AS value").format(
+            sql.Identifier(id_column), sql.Identifier(field)
+        )
+        if field
+        else sql.SQL("*")
+    )
+    filters = []
+    params = []
+    if ids:
+        filters.append(sql.SQL("{}=ANY(%s)").format(sql.Identifier(id_column)))
+        params.append(ids)
+    if run_id:
+        filters.append(sql.SQL("run_id=%s"))
+        params.append(run_id)
+    if start is not None:
+        filters.append(
+            sql.SQL(
+                "time >= %s AND time < %s"
+                if end_exclusive
+                else "time >= %s AND time <= %s"
+            )
+        )
+        params.extend((start, end))
+    else:
+        # Existing point-query semantics use +/- one second.
+        filters.append(
+            sql.SQL("time BETWEEN %s-interval '1 second' AND %s+interval '1 second'")
+        )
+        params.extend((time, time))
+    statement = sql.SQL(
+        "SELECT {} FROM {}.{} WHERE {} ORDER BY time,{} LIMIT %s OFFSET %s"
+    ).format(
+        columns,
+        sql.Identifier(source),
+        sql.Identifier(table),
+        sql.SQL(" AND ").join(filters),
+        sql.Identifier(id_column),
+    )
+    return await read_rows(conn, statement, tuple(params) + (limit, offset))
+
+
+async def sensor_runs(conn, *, run_id=None, limit=100, offset=0):
+    where = " AND r.run_id=%s" if run_id else ""
+    params = (run_id,) if run_id else ()
+    return await read_rows(
+        conn,
+        """SELECT r.run_id,r.name,r.created_by,r.created_at,r.status,
+        result.payload FROM analysis.runs r JOIN LATERAL
+        (SELECT payload FROM analysis.results WHERE run_id=r.run_id
+         AND result_type='sensor_placement' ORDER BY created_at DESC,result_id DESC LIMIT 1) result ON true
+        WHERE r.run_type='sensor_placement'"""
+        + where
+        + " ORDER BY r.created_at DESC,r.run_id LIMIT %s OFFSET %s",
+        params + (limit, offset),
+    )
+
+
+async def history_rows(
+    conn: AsyncConnection, descriptors: list[dict[str, Any]], *,
+    start: datetime, end: datetime, run_id: UUID | None, limit: int, offset: int,
+) -> list[dict[str, Any]]:
+    """Page points across raw, cleaned and model series in one SQL statement."""
+    parts, params = [], []
+    for index, item in enumerate(descriptors):
+        source, field = item["source"], item["field"]
+        if source == "scada":
+            table, id_column, fields = "measurements", "device_id", {"monitored_value", "cleaned_value"}
+        elif source in {"realtime", "analysis"}:
+            node = item["element_type"] == "junction"
+            table, id_column = ("node_results", "node_id") if node else ("link_results", "link_id")
+            fields = {"pressure"} if node else {"flow"}
+        else:
+            raise ValueError("unsupported history source")
+        if field not in fields or (source == "analysis" and run_id is None):
+            raise ValueError("unsupported history field or missing run")
+        parts.append(sql.SQL(
+            "SELECT %s::integer AS series_index,time,{} AS value FROM {}.{} "
+            "WHERE {}=%s AND time >= %s AND time <= %s{}"
+        ).format(
+            sql.Identifier(field), sql.Identifier(source), sql.Identifier(table),
+            sql.Identifier(id_column), sql.SQL(" AND run_id=%s" if source == "analysis" else ""),
+        ))
+        params.extend((index, item["id"], start, end))
+        if source == "analysis":
+            params.append(run_id)
+    if not parts:
+        return []
+    statement = sql.SQL(
+        "SELECT series_index,time,value FROM ({}) points "
+        "ORDER BY series_index, time LIMIT %s OFFSET %s"
+    ).format(sql.SQL(" UNION ALL ").join(parts))
+    return await read_rows(conn, statement, tuple(params) + (limit, offset))
+
+
+async def evidence_page(
+    conn: AsyncConnection, *, project_id: UUID, task_id: UUID, evidence_id: str, data_path: list[str],
+    limit: int, offset: int, mcp_sources: list[str], full_snapshot: bool = False,
+) -> dict[str, Any]:
+    """Project the immutable snapshot in PostgreSQL before transferring JSON."""
+    return await _snapshot_page(conn, stored_sql="""
+        SELECT document->'data_snapshot' AS snapshot
+        FROM workbench.operational_task_evidence
+        WHERE project_id=%s AND task_id=%s AND evidence_id=%s
+          AND jsonb_typeof(document->'data_snapshot')='object'
+        """, stored_params=(project_id, task_id, evidence_id), data_path=data_path,
+        limit=limit, offset=offset, mcp_sources=mcp_sources, full_snapshot=full_snapshot)
+
+
+async def _snapshot_page(
+    conn: AsyncConnection, *, stored_sql: str, stored_params: tuple, data_path: list[str],
+    limit: int, offset: int, mcp_sources: list[str], full_snapshot: bool = False,
+    max_bytes: int = MAX_QUERY_BYTES,
+) -> dict[str, Any]:
+    """Shared JSONB projection; callers supply only fixed, scoped SQL sources."""
+    def descriptor(value, path):
+        return f"""jsonb_build_object(
+            'collection_type',jsonb_typeof({value}),
+            'count',CASE jsonb_typeof({value}) WHEN 'array' THEN jsonb_array_length({value})
+                WHEN 'string' THEN char_length({value} #>> '{{}}')
+                ELSE (SELECT count(*) FROM jsonb_object_keys({value})) END,
+            'data_path',to_jsonb({path}))"""
+
+    collection = descriptor("p.value", "p.path")
+    child = descriptor("child.item", "p.path || ARRAY[child.key]")
+    statement = """
+        WITH stored AS MATERIALIZED (
+            STORED_QUERY
+        ), captured AS (
+            SELECT snapshot,
+                snapshot->>'source'=ANY(%s::text[])
+                    AND jsonb_typeof(snapshot->'data')='object' AS is_mcp
+            FROM stored
+        ), selected AS MATERIALIZED (
+            SELECT snapshot,
+                CASE WHEN is_mcp THEN (snapshot->'data')-'data' END AS query_metadata,
+                (CASE WHEN is_mcp THEN snapshot->'data'->'data'
+                      ELSE snapshot->'data' END) #> %s::text[] AS value
+            FROM captured
+        ), entries AS (
+            SELECT ordinality AS position, NULL::text AS key, item AS value
+            FROM selected, jsonb_array_elements(
+                CASE WHEN jsonb_typeof(value)='array' THEN value ELSE '[]'::jsonb END
+            ) WITH ORDINALITY AS a(item, ordinality)
+            UNION ALL
+            SELECT row_number() OVER (ORDER BY e.key COLLATE "C"), e.key, e.item
+            FROM selected, jsonb_each(
+                CASE WHEN jsonb_typeof(value)='object' THEN value ELSE '{}'::jsonb END
+            ) AS e(key,item)
+        ), paged AS MATERIALIZED (
+            SELECT * FROM entries ORDER BY position LIMIT %s OFFSET %s
+        ), paths AS (
+            SELECT paged.*, jsonb_typeof(selected.value) AS parent_type,
+                %s::text[] || ARRAY[CASE WHEN jsonb_typeof(selected.value)='array'
+                    THEN (position-1)::text ELSE key END] AS path
+            FROM paged,selected
+        ), summaries AS (
+            SELECT p.position,p.key,p.value AS original,
+                CASE WHEN jsonb_typeof(p.value) IN ('array','object') OR
+                    (jsonb_typeof(p.value)='string' AND char_length(p.value #>> '{}')>256) THEN
+                    CASE WHEN p.parent_type='object' OR jsonb_typeof(p.value) IN ('array','string')
+                        THEN COLLECTION_PREVIEW
+                        ELSE COALESCE((SELECT jsonb_object_agg(child.key,
+                            CASE WHEN jsonb_typeof(child.item) IN ('array','object') OR
+                                (jsonb_typeof(child.item)='string' AND char_length(child.item #>> '{}')>256)
+                                THEN CHILD_PREVIEW ELSE child.item END)
+                            FROM jsonb_each(p.value) AS child(key,item)), '{}'::jsonb)
+                    END ELSE p.value END AS value
+            FROM paths p
+        ), projected AS (
+            SELECT CASE WHEN %s THEN snapshot ELSE jsonb_build_object(
+                'snapshot_metadata',snapshot-'data', 'query_metadata',query_metadata,
+                'nested',EXISTS(SELECT 1 FROM summaries WHERE original<>value),
+                'value_type',jsonb_typeof(value),
+                'total',CASE jsonb_typeof(value)
+                    WHEN 'array' THEN jsonb_array_length(value)
+                    WHEN 'object' THEN (SELECT count(*) FROM jsonb_object_keys(value))
+                    WHEN 'string' THEN char_length(value #>> '{}') END,
+                'value',CASE jsonb_typeof(value)
+                    WHEN 'array' THEN COALESCE((SELECT jsonb_agg(value ORDER BY position) FROM summaries),'[]'::jsonb)
+                    WHEN 'object' THEN COALESCE((SELECT jsonb_object_agg(key,value) FROM summaries),'{}'::jsonb)
+                    WHEN 'string' THEN to_jsonb(substring(value #>> '{}' FROM %s FOR %s))
+                    ELSE value END
+                ) END AS payload, value IS NULL AS missing_path
+            FROM selected
+        ), sized AS MATERIALIZED (
+            SELECT payload,missing_path,octet_length(payload::text)>%s AS too_large
+            FROM projected
+        )
+        SELECT CASE WHEN NOT too_large THEN payload END AS payload,too_large,missing_path
+        FROM sized
+    """
+    statement = statement.replace("STORED_QUERY", stored_sql).replace("COLLECTION_PREVIEW", collection).replace("CHILD_PREVIEW", child)
+    rows = await read_rows(conn, statement, (
+        *stored_params, mcp_sources, data_path, limit, offset,
+        data_path, full_snapshot, offset + 1, limit, max_bytes,
+    ))
+    if not rows:
+        raise HTTPException(404, "Stored query document was not found")
+    row = rows[0]
+    if row["too_large"]:
+        raise HTTPException(413, "Query exceeds the byte budget; select a deeper data_path or smaller page")
+    if row.get("missing_path") and not full_snapshot:
+        raise HTTPException(422, "Evidence data_path does not name a stored value")
+    return row["payload"]
+
+
+async def task_context_summary(conn: AsyncConnection, *, project_id: UUID, task_id: UUID, report_revision: int | None) -> dict:
+    """Read fixed-size metadata and explicit previews, never a complete report."""
+    report = """jsonb_build_object('revision',r.revision,
+        'execution_generation',r.document->'execution_generation',
+        'stage',r.document#>'{report,stage}','outcome',r.document#>'{report,outcome}',
+        'current_conclusion_preview',left(r.document#>>'{report,current_conclusion}',200),
+        'preview_only',true,'detail',jsonb_build_object('section','report','report_revision',r.revision))"""
+    rows = await read_rows(conn, """
+        SELECT jsonb_build_object('task',jsonb_build_object(
+            'task_id',t.task_id,'task_type',t.task_type,'workflow',t.document->>'workflow','status',t.status,
+            'lifecycle_status',t.lifecycle_status,'priority',t.priority,'revision',t.revision,
+            'execution_generation',t.execution_generation,
+            'current_report_revision',t.current_report_revision,
+            'title_preview',left(t.document->>'title',160),'preview_only',true,
+            'detail',jsonb_build_object('section','task')),
+            'latest_report',(SELECT REPORT_SUMMARY FROM workbench.operational_task_reports r
+                WHERE r.project_id=t.project_id AND r.task_id=t.task_id ORDER BY r.revision DESC LIMIT 1),
+            'selected_report',CASE WHEN %s::int IS NULL THEN NULL ELSE
+                (SELECT REPORT_SUMMARY FROM workbench.operational_task_reports r
+                 WHERE r.project_id=t.project_id AND r.task_id=t.task_id AND r.revision=%s) END
+            ) AS payload
+        FROM workbench.operational_tasks t WHERE t.project_id=%s AND t.task_id=%s
+        """.replace("REPORT_SUMMARY", report), (report_revision, report_revision, project_id, task_id))
+    if not rows:
+        raise HTTPException(404, "Task was not found in this project")
+    result = rows[0]["payload"]
+    if report_revision is not None and result["selected_report"] is None:
+        raise HTTPException(404, "Selected report revision was not found")
+    return result
+
+
+async def task_context_page(
+    conn: AsyncConnection, *, project_id: UUID, task_id: UUID, section: str,
+    report_revision: int | None, data_path: list[str], limit: int, offset: int,
+) -> dict:
+    """Page only the requested task section inside PostgreSQL, before transfer."""
+    collections = {
+        "evidence": ("operational_task_evidence", "evidence_id"),
+        "actions": ("operational_task_actions", "action_id"),
+        "access_requests": ("operational_task_access_requests", "access_request_id"),
+    }
+    documents = {"task": "operational_tasks", "report": "operational_task_reports", "plan": "operational_task_plans"}
+    if section in {"access_requests", "progress"}:
+        table = "operational_task_access_requests" if section == "access_requests" else "operational_task_progress"
+        available = await read_rows(conn, "SELECT to_regclass(%s) AS relation", ("workbench." + table,))
+        if not available[0]["relation"]:
+            return {"snapshot_metadata": {"section": section, "available": False}, "value": None, "value_type": "null", "total": None, "nested": False}
+    projection_offset = offset
+    if section in collections:
+        table, key = collections[section]
+        # These identifiers come exclusively from the fixed table above.
+        relation = sql.Identifier("workbench", table).as_string()
+        identifier = sql.Identifier(key).as_string()
+        if data_path:
+            stored = f"""SELECT jsonb_build_object('section',%s::text,'data',jsonb_build_object({identifier}::text,document)) AS snapshot
+                FROM {relation} WHERE project_id=%s AND task_id=%s AND {identifier}::text=%s"""
+            params = (section, project_id, task_id, data_path[0])
+        else:
+            # Fetch just this page of rows. Collection keys remain stable across
+            # paging, so every nested data_path uses the real evidence/action ID.
+            stored = f"""WITH items AS MATERIALIZED (
+                SELECT {identifier}::text AS key,document FROM {relation}
+                WHERE project_id=%s AND task_id=%s
+                ORDER BY {identifier}::text COLLATE "C" LIMIT %s OFFSET %s)
+                SELECT jsonb_build_object('section',%s::text,
+                    'total',(SELECT count(*) FROM {relation} WHERE project_id=%s AND task_id=%s),
+                    'data',COALESCE(jsonb_object_agg(key,document),'{{}}'::jsonb)) AS snapshot FROM items"""
+            params = (project_id, task_id, limit + 1, offset, section, project_id, task_id)
+            projection_offset = 0
+    elif section in documents:
+        relation = sql.Identifier("workbench", documents[section]).as_string()
+        where = ""
+        params = (section, project_id, task_id)
+        if section == "report":
+            where = " AND (%s::int IS NULL OR revision=%s) ORDER BY revision DESC LIMIT 1"
+            params += (report_revision, report_revision)
+        elif section == "plan":
+            where = " ORDER BY revision DESC LIMIT 1"
+        stored = f"SELECT jsonb_build_object('section',%s::text,'data',document) AS snapshot FROM {relation} WHERE project_id=%s AND task_id=%s{where}"
+    elif section == "run":
+        stored = """SELECT jsonb_build_object('section','run','data',to_jsonb(r)-'project_id') AS snapshot
+            FROM workbench.operational_task_runs r WHERE project_id=%s AND task_id=%s ORDER BY generation DESC LIMIT 1"""
+        params = (project_id, task_id)
+    elif section == "progress":
+        stored = """SELECT jsonb_build_object('section','progress','data',p.document) AS snapshot
+            FROM workbench.operational_task_progress p
+            JOIN workbench.operational_task_runs r ON r.project_id=p.project_id AND r.task_id=p.task_id AND r.run_id=p.run_id
+            JOIN workbench.operational_tasks t ON t.project_id=p.project_id AND t.task_id=p.task_id AND t.execution_generation=r.generation
+            WHERE p.project_id=%s AND p.task_id=%s"""
+        params = (project_id, task_id)
+    else:
+        raise HTTPException(422, "Unknown task context section")
+    return await _snapshot_page(conn, stored_sql=stored, stored_params=params,
+        data_path=data_path, limit=limit, offset=projection_offset, mcp_sources=[], max_bytes=8000)
+
+
+async def task_data_scopes(conn: AsyncConnection, *, project_id: UUID, task_id: UUID, permissions: set[str]) -> list[str]:
+    """Bound scope discovery to seven known sources, without loading task detail."""
+    known = ["workbench_runtime", "peripheral_points", "peripheral_work_orders", "peripheral_observations", "peripheral_test_data", "scada_devices", "scada_readings"]
+    scopes = {name for name in known if ("scada.view" if name.startswith("scada_") else "simulation.view") in permissions}
+    available = await read_rows(conn, "SELECT to_regclass('workbench.operational_task_access_requests') AS relation")
+    if available[0]["relation"]:
+        rows = await read_rows(conn, """SELECT DISTINCT scope FROM workbench.operational_task_access_requests,
+            LATERAL jsonb_array_elements_text(COALESCE(document->'approved_scopes','[]'::jsonb)) AS granted(scope)
+            WHERE project_id=%s AND task_id=%s AND status='approved' AND scope=ANY(%s::text[])""", (project_id, task_id, known))
+        scopes.update(row["scope"] for row in rows)
+    return [name for name in known if name in scopes]

app/infra/db/postgresql/scada.py

@@ -111,14 +111,16 @@ class ScadaInfoRepository:
         conn: AsyncConnection,
         node_ids: list[str],
         link_ids: list[str],
+        *, row_limit: int | None = None,
     ) -> list[dict[str, Any]]:
         if not node_ids and not link_ids:
             return []
         async with conn.cursor() as cur:
             await cur.execute(
                 _SCADA_VIEW_SELECT
-                + " WHERE node_id = ANY(%s) OR link_id = ANY(%s) ORDER BY device_id",
-                (node_ids, link_ids),
+                + " WHERE node_id = ANY(%s) OR link_id = ANY(%s) ORDER BY device_id"
+                + (" LIMIT %s" if row_limit is not None else ""),
+                (node_ids, link_ids) + ((row_limit,) if row_limit is not None else ()),
             )
             return [_device(record) for record in await cur.fetchall()]
 

app/infra/db/postgresql/spatial_queries.py

@@ -0,0 +1,71 @@
+"""Bounded nearest-model queries on the already authorized project connection."""
+
+from math import asin, cos, degrees, radians, sin
+from typing import Any
+
+from psycopg import AsyncConnection, sql
+
+from app.infra.db.postgresql.read_queries import ELEMENT_VIEWS, read_rows
+
+_MODEL_RELATIONS = {
+    kind: sql.Identifier("gis", view) for kind, view in ELEMENT_VIEWS.items()
+}
+
+
+def _search_boxes(
+    longitude: float, latitude: float, radius_m: float,
+) -> list[tuple[float, float, float, float]]:
+    # A deliberately smaller-than-WGS84 curvature radius gives a conservative
+    # geographic envelope. It is only a GiST candidate filter; geography below
+    # determines the actual radius and distances. Split at the antimeridian.
+    angle = (radius_m + 10) / 6_300_000
+    low, high = latitude - degrees(angle), latitude + degrees(angle)
+    if low < -85 or high > 85:
+        # Avoid Mercator's polar singularity; retain exact geography filtering.
+        return []
+    width = degrees(asin(min(1.0, sin(angle) / cos(radians(latitude)))))
+    left, right = longitude - width, longitude + width
+    if left < -180:
+        return [(-180, low, right, high), (left + 360, low, 180, high)]
+    if right > 180:
+        return [(left, low, 180, high), (-180, low, right - 360, high)]
+    return [(left, low, right, high)]
+
+
+async def nearby_elements(
+    conn: AsyncConnection,
+    element_type: str,
+    *,
+    center: tuple[float, float],
+    radius_m: float,
+    limit: int,
+    offset: int,
+) -> list[dict[str, Any]]:
+    """Rank complete model geometries by spheroidal distance, never endpoints."""
+    relation = _MODEL_RELATIONS[element_type]
+    filters = []
+    parameters: list[Any] = [*center]
+    for box in _search_boxes(*center, radius_m):
+        filters.append(sql.SQL(
+            "v.geom && ST_Transform(ST_MakeEnvelope(%s,%s,%s,%s,4326),3857)"
+        ))
+        parameters.extend(box)
+    candidate_filter = sql.SQL(" OR ").join(filters) if filters else sql.SQL("TRUE")
+    statement = sql.SQL("""
+        WITH target AS (
+            SELECT ST_SetSRID(ST_MakePoint(%s,%s),4326)::geography AS point
+        ), candidates AS MATERIALIZED (
+            SELECT v.id, to_jsonb(v)-'geom' AS properties,
+                   ST_Transform(v.geom,4326)::geography AS shape
+            FROM {} v
+            WHERE v.geom IS NOT NULL AND NOT ST_IsEmpty(v.geom) AND ({})
+        )
+        SELECT c.id AS element_id, c.properties,
+               ST_Distance(c.shape,t.point,true) AS distance_m
+        FROM candidates c CROSS JOIN target t
+        WHERE ST_DWithin(c.shape,t.point,%s,true)
+        ORDER BY distance_m,c.id
+        LIMIT %s OFFSET %s
+    """).format(relation, candidate_filter)
+    parameters.extend((radius_m, limit, offset))
+    return await read_rows(conn, statement, tuple(parameters))