From 1f70c201287f673c1a8e229871283e3af5ded699 Mon Sep 17 00:00:00 2001
From: Luka Skugor <luka.skugor@dfinity.org>
Date: Fri, 6 Nov 2026 10:32:19 +0100
Subject: [PATCH 1/2] factory: make the cockpit fast

Batch the snapshot's N+1 reads (viewer permissions, sub-agents per member,
paused labels, agent rows, roster children), serve a stale snapshot and node
projection while one rebuild runs, cache each viewer's cut with its ETag digest,
render the snapshot off the event loop, read member costs side by side, and
coalesce factory_changed frames in the web to one refresh per factory per 2 s.
Adds a live-size benchmark (dev/benchmarks/factory_cockpit.py) and a query and
latency budget test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 dev/benchmarks/factory_cockpit.py             | 318 ++++++++++++++
 omnigent/server/factory/api/cockpit_read.py   |  41 +-
 omnigent/server/factory/membership.py         |  29 +-
 omnigent/server/factory/nodes.py              | 106 ++++-
 omnigent/server/factory/roster.py             |  58 ++-
 omnigent/server/factory/snapshot.py           | 206 +++++++--
 omnigent/server/factory/tasks.py              |  54 ++-
 .../stores/conversation_store/__init__.py     |  21 +
 .../conversation_store/sqlalchemy_store.py    |  29 ++
 tests/server/factory/perf_world.py            | 406 ++++++++++++++++++
 tests/server/factory/test_cockpit_perf.py     | 130 ++++++
 web/src/hooks/SessionUpdatesProvider.test.tsx |  36 ++
 web/src/hooks/SessionUpdatesProvider.tsx      |  13 +-
 web/src/hooks/useFactory.ts                   |  66 +++
 14 files changed, 1435 insertions(+), 78 deletions(-)
 create mode 100644 dev/benchmarks/factory_cockpit.py
 create mode 100644 tests/server/factory/perf_world.py
 create mode 100644 tests/server/factory/test_cockpit_perf.py

diff --git a/dev/benchmarks/factory_cockpit.py b/dev/benchmarks/factory_cockpit.py
new file mode 100644
index 00000000000..0a406ea62b2
--- /dev/null
+++ b/dev/benchmarks/factory_cockpit.py
@@ -0,0 +1,318 @@
+"""
+The factory cockpit's read latency on a factory the size of a live one.
+
+Seeds a SQLite factory (``tests/server/factory/perf_world.py``: 20 projects,
+~50 members, 5k events, 600 decisions and proposals, 30 demos, 2k chat
+messages), serves the factory routes with uvicorn on a thread, and measures:
+
+- **cold**: each endpoint right after every cache is dropped;
+- **warm**: each endpoint read again and again;
+- **load**: two open cockpit tabs polling as the web does while agents write
+  events (~2/s, each one a write the cockpit must see), with a ``/ping`` probe
+  that shows how long any other request waits for the event loop.
+
+Run::
+
+    OMNIGENT_SKIP_WEB_UI=true uv run python -m dev.benchmarks.factory_cockpit
+    uv run python -m dev.benchmarks.factory_cockpit --profile snapshot   # cProfile one path
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import statistics
+import tempfile
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+ENDPOINTS: dict[str, str] = {
+    "snapshot": "/v1/factory/factories/{fid}",
+    "node_product": "/v1/factory/factories/{fid}/n/product",
+    "node_factory": "/v1/factory/factories/{fid}/n/factory",
+    "needs_you": "/v1/factory/factories/{fid}/needs-you",
+    "chat": "/v1/factory/factories/{fid}/chat?thread=factory",
+    "status_summary": "/v1/factory/factories/{fid}/status-summary",
+    "preflight": "/v1/factory/preflight",
+    "levels": "/v1/factory/factories/{fid}/levels",
+    "decisions": "/v1/factory/decisions",
+    "events": "/v1/factory/factories/{fid}/events",
+    "factories": "/v1/factory/factories",
+    "definition": "/v1/factory/factories/{fid}/definition",
+    "people": "/v1/factory/factories/{fid}/people",
+}
+
+POLLERS: dict[str, float] = {
+    "snapshot": 5.0,
+    "node_product": 5.0,
+    "decisions": 10.0,
+    "events": 5.0,
+    "factories": 15.0,
+    "definition": 30.0,
+    "people": 30.0,
+    "needs_you": 10.0,
+    "chat": 5.0,
+}
+"""Endpoint -> seconds between a tab's reads (what the cockpit polls, before this change)."""
+
+
+class Queries:
+    """Counts SQL statements (and adds a round trip's latency to each, like Postgres)."""
+
+    count = 0
+    latency_s = 0.0
+    _lock = threading.Lock()
+
+    @classmethod
+    def install(cls, db_uri: str, latency_ms: float) -> None:
+        from sqlalchemy import event
+
+        from omnigent.db.utils import get_or_create_engine
+
+        cls.latency_s = latency_ms / 1000
+
+        def before(_conn: Any, _cursor: Any, statement: str, *_args: Any) -> None:
+            if statement.startswith("PRAGMA"):
+                return  # SQLite's per-connection setup: no round trip on Postgres
+            with cls._lock:
+                cls.count += 1
+            if cls.latency_s:
+                time.sleep(cls.latency_s)
+
+        event.listen(get_or_create_engine(db_uri), "before_cursor_execute", before)
+
+
+def reset_caches() -> None:
+    """Drop every cockpit read cache in this process."""
+    from omnigent.server.factory import nodes, snapshot, status
+
+    snapshot.reset_cache()
+    nodes.reset_cache()
+    status.reset()
+    reset = getattr(nodes, "reset_names", None)
+    if reset is not None:
+        reset()
+
+
+def pct(values: list[float], p: float) -> float:
+    if not values:
+        return 0.0
+    ordered = sorted(values)
+    k = min(len(ordered) - 1, max(0, round(p / 100 * (len(ordered) - 1))))
+    return ordered[k]
+
+
+def summary(values: list[float]) -> dict[str, float]:
+    return {
+        "n": len(values),
+        "p50": round(pct(values, 50) * 1000, 1),
+        "p95": round(pct(values, 95) * 1000, 1),
+        "max": round(max(values, default=0) * 1000, 1),
+        "mean": round(statistics.fmean(values) * 1000, 1) if values else 0.0,
+    }
+
+
+class Served:
+    """The app on a uvicorn thread."""
+
+    def __init__(self, app: Any) -> None:
+        import socket
+
+        import uvicorn
+
+        sock = socket.socket()
+        sock.bind(("127.0.0.1", 0))
+        self.port = sock.getsockname()[1]
+        sock.close()
+        config = uvicorn.Config(app, host="127.0.0.1", port=self.port, log_level="warning")
+        self.server = uvicorn.Server(config)
+        self.thread = threading.Thread(target=self.server.run, daemon=True)
+        self.thread.start()
+        deadline = time.monotonic() + 20
+        while not self.server.started:
+            if time.monotonic() > deadline:
+                raise RuntimeError("uvicorn did not start")
+            time.sleep(0.05)
+        self.base_url = f"http://127.0.0.1:{self.port}"
+
+    def stop(self) -> None:
+        self.server.should_exit = True
+        self.thread.join(timeout=10)
+
+
+def build(scale_name: str, tmp: Path) -> tuple[Any, str, str]:
+    """:returns: ``(app, factory id, db uri)`` of a freshly seeded factory."""
+    from tests.server.factory import perf_world
+
+    db_uri = f"sqlite:///{tmp / 'bench.db'}"
+    app = perf_world.build_app(db_uri, tmp)
+    scale = perf_world.SMALL if scale_name == "small" else perf_world.Scale()
+    started = time.monotonic()
+    fid = perf_world.seed_live_factory(db_uri, app, scale)
+    print(f"seeded {scale_name} factory {fid} in {time.monotonic() - started:.1f}s")
+    _add_bench_routes(app, fid)
+    return app, fid, db_uri
+
+
+def _add_bench_routes(app: Any, fid: str) -> None:
+    from omnigent.server.factory import events
+
+    @app.post("/bench/reset")
+    async def bench_reset() -> dict[str, bool]:
+        reset_caches()
+        return {"ok": True}
+
+    @app.post("/bench/event")
+    async def bench_event() -> dict[str, bool]:
+        # What an agent's report does: one recorded event the cockpit must see.
+        await events.emit(
+            app.state, fid, "task.state", project="assistant-core", summary="bench write"
+        )
+        return {"ok": True}
+
+
+async def _get(client: httpx.AsyncClient, path: str, headers: dict[str, str]) -> float:
+    started = time.perf_counter()
+    resp = await client.get(path, headers=headers)
+    took = time.perf_counter() - started
+    if resp.status_code not in (200, 304):
+        raise RuntimeError(f"GET {path}: {resp.status_code} {resp.text[:300]}")
+    return took
+
+
+async def measure(base_url: str, fid: str, *, warm_n: int, load_s: float) -> dict[str, Any]:
+    from tests.server.factory.perf_world import OWNER
+
+    headers = {"X-Forwarded-Email": OWNER}
+    out: dict[str, Any] = {"cold": {}, "warm": {}, "load": {}, "sizes": {}}
+    limits = httpx.Limits(max_connections=64)
+    async with httpx.AsyncClient(base_url=base_url, timeout=120, limits=limits) as client:
+        for name, template in ENDPOINTS.items():
+            path = template.format(fid=fid)
+            await client.post("/bench/reset")
+            before = Queries.count
+            out["cold"][name] = round(await _get(client, path, headers) * 1000, 1)
+            out.setdefault("queries", {})[name] = Queries.count - before
+            resp = await client.get(path, headers=headers)
+            out["sizes"][name] = len(resp.content)
+            times = [await _get(client, path, headers) for _ in range(warm_n)]
+            out["warm"][name] = summary(times)
+            print(
+                f"  {name:16s} cold {out['cold'][name]:8.1f} ms "
+                f"({out['queries'][name]:5d} q, {out['sizes'][name] // 1024:4d} KB)"
+                f"   warm {out['warm'][name]}"
+            )
+        if load_s > 0:
+            out["load"] = await _load(client, fid, headers, load_s)
+    return out
+
+
+async def _load(
+    client: httpx.AsyncClient, fid: str, headers: dict[str, str], seconds: float
+) -> dict[str, Any]:
+    """Two tabs polling, agents writing ~2 events/s, and a /ping probe at 10/s."""
+    stop = time.monotonic() + seconds
+    times: dict[str, list[float]] = {name: [] for name in POLLERS}
+    pings: list[float] = []
+    writes: list[float] = []
+    etags: dict[tuple[int, str], str] = {}
+
+    async def poller(tab: int, name: str, every: float) -> None:
+        path = ENDPOINTS[name].format(fid=fid)
+        await asyncio.sleep(every * tab / 2)
+        while time.monotonic() < stop:
+            extra = dict(headers)
+            if (tab, name) in etags:
+                extra["If-None-Match"] = etags[(tab, name)]
+            started = time.perf_counter()
+            resp = await client.get(path, headers=extra)
+            times[name].append(time.perf_counter() - started)
+            if tag := resp.headers.get("etag"):
+                etags[(tab, name)] = tag
+            await asyncio.sleep(every)
+
+    async def writer() -> None:
+        while time.monotonic() < stop:
+            started = time.perf_counter()
+            await client.post("/bench/event")
+            writes.append(time.perf_counter() - started)
+            await asyncio.sleep(0.5)
+
+    async def pinger() -> None:
+        while time.monotonic() < stop:
+            started = time.perf_counter()
+            await client.get("/ping")
+            pings.append(time.perf_counter() - started)
+            await asyncio.sleep(0.1)
+
+    tasks = [poller(tab, name, every) for tab in (0, 1) for name, every in POLLERS.items()]
+    await asyncio.gather(*tasks, writer(), pinger())
+    result = {name: summary(values) for name, values in times.items()}
+    result["agent_write"] = summary(writes)
+    result["ping"] = summary(pings)
+    for name, row in result.items():
+        print(f"  load {name:14s} {row}")
+    return result
+
+
+def profile(app: Any, fid: str, which: str, out: Path) -> None:
+    """cProfile one read path in process (cold caches) and print the top of it."""
+    import cProfile
+    import pstats
+
+    from fastapi.testclient import TestClient
+
+    from tests.server.factory.perf_world import OWNER
+
+    path = ENDPOINTS[which].format(fid=fid)
+    client = TestClient(app)
+    client.get(path, headers={"X-Forwarded-Email": OWNER})
+    reset_caches()
+    prof = cProfile.Profile()
+    prof.enable()
+    resp = client.get(path, headers={"X-Forwarded-Email": OWNER})
+    prof.disable()
+    print(f"{which}: {resp.status_code}, {len(resp.content)} bytes")
+    prof.dump_stats(str(out))
+    stats = pstats.Stats(prof)
+    stats.sort_stats("cumulative").print_stats(60)
+
+
+def main(argv: list[str] | None = None) -> None:
+    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    parser.add_argument("--scale", choices=["live", "small"], default="live")
+    parser.add_argument("--warm", type=int, default=15, help="warm reads per endpoint")
+    parser.add_argument("--load", type=float, default=30.0, help="seconds of mixed load")
+    parser.add_argument("--profile", choices=list(ENDPOINTS), help="cProfile one endpoint")
+    parser.add_argument("--json", type=Path, help="write the numbers here")
+    parser.add_argument(
+        "--db-latency-ms", type=float, default=1.0, help="added to each SQL statement"
+    )
+    args = parser.parse_args(argv)
+    import logging
+
+    logging.basicConfig(level=logging.ERROR)
+    with tempfile.TemporaryDirectory(prefix="factory-bench-") as tmp:
+        app, fid, db_uri = build(args.scale, Path(tmp))
+        Queries.install(db_uri, args.db_latency_ms)
+        if args.profile:
+            profile(app, fid, args.profile, Path(tmp) / "out.prof")
+            return
+        served = Served(app)
+        try:
+            numbers = asyncio.run(
+                measure(served.base_url, fid, warm_n=args.warm, load_s=args.load)
+            )
+        finally:
+            served.stop()
+    if args.json:
+        args.json.write_text(json.dumps(numbers, indent=2))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/omnigent/server/factory/api/cockpit_read.py b/omnigent/server/factory/api/cockpit_read.py
index 92f5f65a682..0a2c4397c77 100644
--- a/omnigent/server/factory/api/cockpit_read.py
+++ b/omnigent/server/factory/api/cockpit_read.py
@@ -10,11 +10,12 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
 from typing import Any, Final
 
 from fastapi import APIRouter, Query, Request
-from fastapi.responses import JSONResponse, Response
+from fastapi.responses import Response
 
 from omnigent.entities import Conversation
 from omnigent.errors import ErrorCode, OmnigentError
@@ -30,15 +31,15 @@
 from omnigent.server.factory.snapshot import (
     bare_id,
     decision_view,
-    filter_snapshot,
+    filtered_view,
     get_snapshot,
     is_factory_root,
     list_factory_roots,
     project_demo_view,
     readable_for,
     readable_set,
-    snapshot_etag,
     summary_from,
+    view_etag,
 )
 from omnigent.server.factory.start import max_projects
 from omnigent.server.routes._auth_helpers import require_access, require_user
@@ -76,18 +77,24 @@ async def _root(user_id: str | None, factory_id: str, deps: FactoryDeps) -> Conv
 
 async def _view(
     request: Request, user_id: str | None, root: Conversation, *, viewer: bool = True
-) -> dict[str, Any]:
+) -> tuple[dict[str, Any], str]:
+    """:returns: The snapshot as this viewer may see it (a shallow copy) and its weak ETag."""
     deps = factory_deps(request.app.state)
     raw = await get_snapshot(request.app.state, root)
     readable = await readable_for(user_id, raw, deps)
-    body = filter_snapshot(raw, readable)
+    shared, digest = await asyncio.to_thread(filtered_view, raw, readable)
+    body = dict(shared)
+    extras: dict[str, Any] = {}
     if viewer:
         # Per viewer and never cached: role, levers, roster size and who is here.
         block, count = await asyncio.to_thread(_viewer_fields, deps, root.id, user_id)
-        body["viewer"] = block
-        body["people_count"] = count
-        body["presence"] = entries_for(root.id, raw.top_level_of, readable)
-    return body
+        extras = {
+            "viewer": block,
+            "people_count": count,
+            "presence": entries_for(root.id, raw.top_level_of, readable),
+        }
+        body.update(extras)
+    return body, view_etag(digest, extras)
 
 
 def _viewer_fields(
@@ -106,6 +113,13 @@ def _etag_matches(header: str | None, etag: str) -> bool:
     return any(tag.strip().removeprefix("W/") == wanted for tag in header.split(","))
 
 
+def _render(body: Any) -> bytes:
+    """``body`` as :class:`JSONResponse` renders it."""
+    return json.dumps(
+        body, ensure_ascii=False, allow_nan=False, indent=None, separators=(",", ":")
+    ).encode("utf-8")
+
+
 def _factory_name(root: Conversation) -> str:
     return root.labels.get(LABEL_NAME) or display_title(root.title, root.id)
 
@@ -165,7 +179,7 @@ async def list_factories(request: Request) -> dict[str, Any]:
         if root.id not in readable:
             continue
         try:
-            body = await _view(request, user_id, root, viewer=False)
+            body, _ = await _view(request, user_id, root, viewer=False)
             extras = await asyncio.to_thread(_summary_extras, deps, root.id, user_id, body)
             summaries.append({**summary_from(body), **extras})
         except Exception:  # one broken factory must not hide the others
@@ -201,12 +215,13 @@ async def get_factory(request: Request, factory_id: str) -> Response:
     deps = factory_deps(request.app.state)
     user_id = require_user(request, deps.auth_provider)
     root = await _root(user_id, factory_id, deps)
-    body = await _view(request, user_id, root)
-    etag = snapshot_etag(body)
+    body, etag = await _view(request, user_id, root)
     headers = {"ETag": etag, "Cache-Control": "no-cache"}
     if _etag_matches(request.headers.get("if-none-match"), etag):
         return Response(status_code=304, headers=headers)
-    return JSONResponse(content=body, headers=headers)
+    # ~0.5 MB on a big factory: rendered off the event loop.
+    content = await asyncio.to_thread(_render, body)
+    return Response(content=content, media_type="application/json", headers=headers)
 
 
 def _event_view(event: Any, titles: dict[str, str], readable: set[str]) -> dict[str, Any]:
diff --git a/omnigent/server/factory/membership.py b/omnigent/server/factory/membership.py
index 1b4ca32a6f9..7ff7fe2bcbe 100644
--- a/omnigent/server/factory/membership.py
+++ b/omnigent/server/factory/membership.py
@@ -9,7 +9,10 @@
 
 from __future__ import annotations
 
+import contextlib
+import contextvars
 import logging
+from collections.abc import Iterator
 from dataclasses import asdict, dataclass
 from typing import TYPE_CHECKING, Any
 
@@ -142,6 +145,30 @@ def resolve_factory_id(conv: Conversation, store: ConversationStore) -> str | No
     return root.labels.get(LABEL_ID) or None
 
 
+_agent_rows: contextvars.ContextVar[dict[str, Any] | None] = contextvars.ContextVar(
+    "factory_agent_rows", default=None
+)
+
+
+@contextlib.contextmanager
+def agent_rows_memo() -> Iterator[None]:
+    """Read each agent row once while one view of a factory is built (its worker threads too)."""
+    token = _agent_rows.set({})
+    try:
+        yield
+    finally:
+        _agent_rows.reset(token)
+
+
+def _agent_row(deps: FactoryDeps, agent_id: str) -> Any:
+    memo = _agent_rows.get()
+    if memo is None:
+        return deps.agent_store.get(agent_id)
+    if agent_id not in memo:
+        memo[agent_id] = deps.agent_store.get(agent_id)
+    return memo[agent_id]
+
+
 def load_session_spec(conv: Conversation, deps: FactoryDeps) -> AgentSpec | None:
     """
     Load the spec *conv* runs: its agent's, or the named sub-agent within it.
@@ -152,7 +179,7 @@ def load_session_spec(conv: Conversation, deps: FactoryDeps) -> AgentSpec | None
     """
     if conv.agent_id is None or deps.agent_cache is None:
         return None
-    agent = deps.agent_store.get(conv.agent_id)
+    agent = _agent_row(deps, conv.agent_id)
     if agent is None:
         return None
     try:
diff --git a/omnigent/server/factory/nodes.py b/omnigent/server/factory/nodes.py
index ed6d501fb6d..62573af811d 100644
--- a/omnigent/server/factory/nodes.py
+++ b/omnigent/server/factory/nodes.py
@@ -127,6 +127,8 @@
 
 FACTS_TTL_S: Final = 5.0
 """How long one factory's projection is reused (any recorded write drops it sooner)."""
+FACTS_STALE_S: Final = 60.0
+"""A projection older than the TTL but younger than this is served while one rebuild runs."""
 BURN_WINDOW_S: Final = 3600
 """Burn is the spend of the last hour."""
 TURNS_LOOKBACK_S: Final = 3 * 3600
@@ -3625,6 +3627,11 @@ def assert_levels(tree: Tree) -> Tree:
 _trees: dict[tuple[str, str], tuple[float, Tree]] = {}
 _locks: dict[tuple[int, str], asyncio.Lock] = {}
 _TREES_MAX: Final = 64
+_generation: dict[str, int] = {}
+"""Factory -> projections dropped so far (a build started before a drop is not kept)."""
+_refreshes: dict[tuple[int, str, str], asyncio.Task[None]] = {}
+_epoch: list[int] = [0]
+"""Bumped by :func:`reset_cache`: no build started before it is kept."""
 
 
 def viewer_key(readable: Iterable[str]) -> str:
@@ -3652,58 +3659,113 @@ async def tree_for(
     The factory as a viewer who may read the sessions *readable* sees it, built
     from the cached snapshot cut to that viewer (``snapshot.filter_snapshot``:
     members, team chips, cost and burn, the base env) and from the store rows
-    of the projects they may read; reused for :data:`FACTS_TTL_S`.
+    of the projects they may read; reused for :data:`FACTS_TTL_S`, then served
+    while one rebuild runs (up to :data:`FACTS_STALE_S`). A recorded write
+    drops it: the next read waits for a fresh one.
 
     :param app_state: The server's state, read on a rebuild for the auto-merge
         targets the server merges into now (a ready PR into one waits on the
         server, not a person); ``None``: a person merges every PR.
     """
-    from omnigent.server.factory.snapshot import filter_snapshot
-
     fid = norm_id(root.id)
     sessions = set(readable)
     key = (fid, viewer_key(sessions))
     cached = _trees.get(key)
-    if cached is not None and time.monotonic() - cached[0] < FACTS_TTL_S:
-        return cached[1]
+    if cached is not None:
+        age = time.monotonic() - cached[0]
+        if age < FACTS_TTL_S:
+            return cached[1]
+        if age < FACTS_STALE_S:
+            _rebuild_in_background(deps, root, raw, sessions, app_state, key)
+            return cached[1]
     lock = _locks.setdefault((id(asyncio.get_running_loop()), fid), asyncio.Lock())
     async with lock:
         cached = _trees.get(key)
         if cached is not None and time.monotonic() - cached[0] < FACTS_TTL_S:
             return cached[1]
+        return await _build_and_keep(deps, root, raw, sessions, app_state, key)
 
-        def build() -> Tree:
-            body = filter_snapshot(raw, sessions)
-            targets = waits.auto_targets(app_state, deps, root.id, lineage=False)
-            facts = load_facts(
-                deps, root, body, readable=sessions, leads=raw.project_pl, targets=targets
-            )
-            tree = Tree(facts, hidden=hidden_projects(raw, sessions))
-            if bad := check_levels(tree):
-                _logger.warning(
-                    "factory %s: the level projection breaks the rule: %s", fid, bad[:5]
-                )
-            return tree
 
-        tree = await asyncio.to_thread(build)
+async def _build_and_keep(
+    deps: FactoryDeps,
+    root: Conversation,
+    raw: RawSnapshot,
+    sessions: set[str],
+    app_state: Any,
+    key: tuple[str, str],
+) -> Tree:
+    from omnigent.server.factory.snapshot import filter_snapshot
+
+    fid = key[0]
+    generation = (_epoch[0], _generation.get(fid, 0))
+
+    def build() -> Tree:
+        body = filter_snapshot(raw, sessions)
+        targets = waits.auto_targets(app_state, deps, root.id, lineage=False)
+        facts = load_facts(
+            deps, root, body, readable=sessions, leads=raw.project_pl, targets=targets
+        )
+        tree = Tree(facts, hidden=hidden_projects(raw, sessions))
+        if bad := check_levels(tree):
+            _logger.warning("factory %s: the level projection breaks the rule: %s", fid, bad[:5])
+        return tree
+
+    tree = await asyncio.to_thread(build)
+    if (_epoch[0], _generation.get(fid, 0)) == generation:
         if len(_trees) >= _TREES_MAX:
             _trees.clear()
         _trees[key] = (time.monotonic(), tree)
-        # Raw asks, machine labels and long titles get the factory's short names
-        # in the background; the page shows their first words until then.
-        node_names.schedule(app_state, deps, root, tree.unnamed)
-        return tree
+    # Raw asks, machine labels and long titles get the factory's short names
+    # in the background; the page shows their first words until then.
+    node_names.schedule(app_state, deps, root, tree.unnamed)
+    return tree
+
+
+def _rebuild_in_background(
+    deps: FactoryDeps,
+    root: Conversation,
+    raw: RawSnapshot,
+    sessions: set[str],
+    app_state: Any,
+    key: tuple[str, str],
+) -> None:
+    """Start one rebuild of the projection *key* unless one runs already (on this loop)."""
+    loop = asyncio.get_running_loop()
+    task_key = (id(loop), *key)
+    running = _refreshes.get(task_key)
+    if running is not None and not running.done():
+        return
+
+    async def refresh() -> None:
+        lock = _locks.setdefault((id(loop), key[0]), asyncio.Lock())
+        try:
+            async with lock:
+                cached = _trees.get(key)
+                if cached is not None and time.monotonic() - cached[0] < FACTS_TTL_S:
+                    return
+                await _build_and_keep(deps, root, raw, sessions, app_state, key)
+        except Exception:  # noqa: BLE001 - the next read builds it in the open
+            _logger.warning("factory %s: background projection failed", key[0], exc_info=True)
+        finally:
+            if _refreshes.get(task_key) is task:
+                _refreshes.pop(task_key, None)
+
+    task = loop.create_task(refresh(), name=f"factory-nodes-{key[0]}")
+    _refreshes[task_key] = task
 
 
 def invalidate(factory_id: str) -> None:
     """Forget a factory's projections (a write changed what they read)."""
     fid = _bare(factory_id)
+    if fid is not None:
+        _generation[fid] = _generation.get(fid, 0) + 1
     for key in [k for k in _trees if k[0] == fid]:
         _trees.pop(key, None)
 
 
 def reset_cache() -> None:
     """Forget every projection (tests)."""
+    _epoch[0] += 1
     _trees.clear()
     _locks.clear()
 
diff --git a/omnigent/server/factory/roster.py b/omnigent/server/factory/roster.py
index 6febcfe9cdd..9a62ac9cf15 100644
--- a/omnigent/server/factory/roster.py
+++ b/omnigent/server/factory/roster.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any
 
 from omnigent.factory import (
@@ -18,14 +19,18 @@
 from omnigent.server.factory.context import FactoryDeps
 from omnigent.server.factory.delivery import member_last_error
 from omnigent.server.factory.membership import load_session_spec, sub_agent_role
-from omnigent.server.routes._sessions.helpers import _session_status_from_cache
+from omnigent.server.routes._sessions.helpers import (
+    _child_session_summary_from_conversation,
+    _latest_message_preview,
+    _session_status_from_cache,
+)
 from omnigent.server.routes._sessions.orchestration import (
     _child_session_summaries_from_conversations,
 )
 from omnigent.stores.conversation_store import runner_seen_is_fresh
 
 if TYPE_CHECKING:
-    from omnigent.entities import Conversation
+    from omnigent.entities import Conversation, ConversationItem
     from omnigent.server.schemas import ChildSessionSummary
 
 # Sub-agents listed per member; a PL runs well under this many.
@@ -88,6 +93,44 @@ async def _children(
     roles = await asyncio.to_thread(
         lambda: {c.id: sub_agent_role(load_session_spec(c, deps)) for c in children}
     )
+    return _child_rows(children, summaries, roles)
+
+
+async def _known_children(
+    members: list[Conversation],
+    known: Mapping[str, list[Conversation]],
+    deps: FactoryDeps,
+) -> list[list[dict[str, Any]]]:
+    """Every member's child rows from children already read: one message read for all."""
+    store = deps.conversation_store
+    flat = [(m.id, c) for m in members for c in known.get(m.id, [])]
+    if not flat:
+        return [[] for _ in members]
+
+    def read() -> tuple[dict[str, list[ConversationItem]], dict[str, str]]:
+        items = store.list_latest_message_items_for_conversations([c.id for _, c in flat], 10)
+        roles = {c.id: sub_agent_role(load_session_spec(c, deps)) for _, c in flat}
+        return items, roles
+
+    items, roles = await asyncio.to_thread(read)
+    out: list[list[dict[str, Any]]] = []
+    for member in members:
+        kids = known.get(member.id, [])
+        summaries = [
+            _child_session_summary_from_conversation(
+                c, member.id, _latest_message_preview(items.get(c.id, []))
+            )
+            for c in kids
+        ]
+        out.append(_child_rows(kids, summaries, roles) if kids else [])
+    return out
+
+
+def _child_rows(
+    children: list[Conversation],
+    summaries: list[ChildSessionSummary],
+    roles: Mapping[str, str],
+) -> list[dict[str, Any]]:
     return [
         {
             "session_id": child.id,
@@ -112,6 +155,7 @@ async def build_roster(
     app_state: Any,
     *,
     humans: bool = False,
+    children: Mapping[str, list[Conversation]] | None = None,
 ) -> dict[str, Any]:
     """
     Build the ``GET /v1/factory/roster`` body.
@@ -121,9 +165,11 @@ async def build_roster(
     :param deps: Factory stores.
     :param app_state: ``app.state`` (runner tunnel registry).
     :param humans: Add ``humans``: the people on the factory roster.
+    :param children: Each member's sub-agents, already read (oldest first);
+        ``None`` reads the oldest 200 of each.
     :returns: ``{factory_id, members: [{…, children: [{…}]}], humans?: [{…}]}``.
     """
-    body = await _members_body(factory_id, members, deps, app_state)
+    body = await _members_body(factory_id, members, deps, app_state, children)
     if humans:
         from omnigent.server.factory.people import roster_humans
 
@@ -136,6 +182,7 @@ async def _members_body(
     members: list[Conversation],
     deps: FactoryDeps,
     app_state: Any,
+    known: Mapping[str, list[Conversation]] | None = None,
 ) -> dict[str, Any]:
     store = deps.conversation_store
     if not members:
@@ -143,7 +190,10 @@ async def _members_body(
     summaries = await _child_session_summaries_from_conversations(members, factory_id, store)
     connectivity = await asyncio.to_thread(store.get_session_connectivity, [m.id for m in members])
     tunnel_registry = getattr(app_state, "tunnel_registry", None)
-    children = await asyncio.gather(*(_children(m, deps) for m in members))
+    if known is not None:
+        children = await _known_children(members, known, deps)
+    else:
+        children = await asyncio.gather(*(_children(m, deps) for m in members))
     rows: list[dict[str, Any]] = []
     for member, summary, kids in zip(members, summaries, children, strict=True):
         labels = member.labels
diff --git a/omnigent/server/factory/snapshot.py b/omnigent/server/factory/snapshot.py
index 33e3dced252..518280825b8 100644
--- a/omnigent/server/factory/snapshot.py
+++ b/omnigent/server/factory/snapshot.py
@@ -16,14 +16,17 @@
 from __future__ import annotations
 
 import asyncio
+import contextvars
 import copy
 import functools
 import hashlib
 import json
 import logging
 import re
+import threading
 import time
 from collections.abc import Iterable, Mapping, Sequence
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Final
 
@@ -59,6 +62,7 @@
     STATE_ACTIVE,
     STATES,
 )
+from omnigent.server.auth import LEVEL_READ, RESERVED_USER_PUBLIC
 from omnigent.server.factory import (
     activity_now,
     auto_merge,
@@ -88,7 +92,11 @@
 from omnigent.server.factory.gating import unmet_dependencies
 from omnigent.server.factory.issues import counts_open, visible_to
 from omnigent.server.factory.launch import launch_drift
-from omnigent.server.factory.membership import factory_members, load_session_spec
+from omnigent.server.factory.membership import (
+    agent_rows_memo,
+    factory_members,
+    load_session_spec,
+)
 from omnigent.server.factory.notify import parse_iso
 from omnigent.server.factory.roster import build_roster, session_url
 from omnigent.server.factory.slackdm import rules as slack_rules
@@ -115,7 +123,11 @@
 _logger = logging.getLogger(__name__)
 
 SNAPSHOT_TTL_S: Final = 2.0
+SNAPSHOT_STALE_S: Final = 60.0
+"""A snapshot older than the TTL but younger than this is served while one rebuild runs."""
 COST_TTL_S: Final = 60.0
+_COST_READERS: Final = 4
+"""Members whose spend is read at once on a cold cache (each read is a few queries)."""
 MAIL_WINDOW_S: Final = 6 * 3600
 NOTIFIED_WINDOW_S: Final = 30 * 86400
 """How far back the snapshot reads Slack DM receipts (``notify.slack``)."""
@@ -124,6 +136,8 @@
 CONTROL_WINDOW_S: Final = 7 * 86400
 ANSWERED_DECISIONS: Final = 10
 PL_TITLE_PREFIX: Final = "PL · "
+_MAX_PARENT_HOPS: Final = 16
+"""Sub-agent -> parent hops :func:`readable_set` follows to a top-level session."""
 
 SANDBOX_STAGE_LABELS: Final = {
     "provisioning": "Provisioning sandbox",
@@ -255,6 +269,25 @@ def subtree_cost(conv: Conversation, store: ConversationStore) -> float | None:
     return value
 
 
+def _subtree_costs(
+    members: Sequence[Conversation], store: ConversationStore
+) -> dict[str, float | None]:
+    """:func:`subtree_cost` of each member; the uncached ones read side by side."""
+    now = time.monotonic()
+    stale = [
+        m for m in members if (hit := _cost_cache.get(m.id)) is None or now - hit[0] >= COST_TTL_S
+    ]
+    if len(stale) > 1:
+        with ThreadPoolExecutor(max_workers=min(_COST_READERS, len(stale))) as pool:
+            # Each read runs in this thread's context (the workspace scope, for one).
+            reads = [
+                pool.submit(contextvars.copy_context().run, subtree_cost, m, store) for m in stale
+            ]
+            for read in reads:
+                read.result()
+    return {m.id: subtree_cost(m, store) for m in members}
+
+
 def _own_cost(conv: Conversation) -> float | None:
     raw = (conv.session_usage or {}).get("total_cost_usd")
     return float(raw) if isinstance(raw, int | float) else None
@@ -719,13 +752,8 @@ def _load(deps: FactoryDeps, factory_id: str, now: float, *, merger: bool = Fals
     )
     members = [m for m in everyone if not is_session_closed(m.labels, m.title)]
     closed = [m for m in everyone if is_session_closed(m.labels, m.title)]
-    children: dict[str, list[Conversation]] = {}
-    for member in members:
-        # The newest 200 (re-dispatches must not fall off the page), shown oldest first.
-        page = store.list_conversations(
-            limit=200, kind="sub_agent", parent_conversation_id=member.id, order="desc"
-        )
-        children[member.id] = list(reversed(page.data))
+    # The newest 200 of each (re-dispatches must not fall off the page), shown oldest first.
+    children = store.list_sub_agents_by_parent([m.id for m in members], 200)
     now_ms = int(now * 1000)
     open_decisions = fstore.list_decisions(factory_id=factory_id, status="open", limit=500)
     # One read of the open proposals: deadlines, Steer's counts and Needs you share it.
@@ -771,7 +799,7 @@ def _load(deps: FactoryDeps, factory_id: str, now: float, *, merger: bool = Fals
         ),
         harness=harness,
         drift={m.id: launch_drift(m, deps) for m in members},
-        costs={m.id: subtree_cost(m, store) for m in members},
+        costs=_subtree_costs(members, store),
         proposals_pending=len(pending),
         # Steer's count: changes to how the factory works (plans and bets wait in Needs you).
         definition_proposals_pending=sum(1 for p in pending if p.kind != "projects"),
@@ -1407,7 +1435,9 @@ async def build_snapshot(app_state: Any, root: Conversation) -> RawSnapshot:
         functools.partial(_load, merger=auto_merge.merger_on(app_state)), deps, factory_id, now
     )
     loaded.now = activity_now.now_states(factory_id)
-    roster = await build_roster(factory_id, loaded.members, deps, app_state)
+    roster = await build_roster(
+        factory_id, loaded.members, deps, app_state, children=loaded.children
+    )
     rows = {r["session_id"]: r for r in roster["members"]}
     child_rows = {c["session_id"]: c for r in roster["members"] for c in r.get("children", [])}
 
@@ -1660,38 +1690,127 @@ def _totals(
 _locks: dict[tuple[int, str], asyncio.Lock] = {}
 _viewers: dict[tuple[str | None, str], tuple[float, set[str]]] = {}
 _VIEWER_CACHE_MAX: Final = 1000
+_generation: dict[str, int] = {}
+"""Factory -> how many times it was forgotten (a build started before a forget is not kept)."""
+_refreshes: dict[tuple[int, str], asyncio.Task[None]] = {}
+_epoch: list[int] = [0]
+"""Bumped by :func:`reset_cache`: no build started before it is kept."""
 
 
 async def get_snapshot(app_state: Any, root: Conversation) -> RawSnapshot:
-    """The unfiltered snapshot of *root*'s factory, rebuilt at most every 2 s."""
+    """
+    The unfiltered snapshot of *root*'s factory, rebuilt at most every 2 s.
+
+    A snapshot older than that but younger than :data:`SNAPSHOT_STALE_S` is
+    answered at once while one rebuild runs in the background; after
+    :func:`forget` (or with none cached) the read waits for a fresh build.
+    """
     key = bare_id(root.id)
     cached = _snapshots.get(key)
-    if cached is not None and time.monotonic() - cached.built_at < SNAPSHOT_TTL_S:
-        return cached
+    if cached is not None:
+        age = time.monotonic() - cached.built_at
+        if age < SNAPSHOT_TTL_S:
+            return cached
+        if age < SNAPSHOT_STALE_S:
+            _refresh_in_background(app_state, root, key)
+            return cached
     lock = _locks.setdefault((id(asyncio.get_running_loop()), key), asyncio.Lock())
     async with lock:
         cached = _snapshots.get(key)
         if cached is not None and time.monotonic() - cached.built_at < SNAPSHOT_TTL_S:
             return cached
+        return await _build_and_keep(app_state, root, key)
+
+
+def _generation_of(key: str) -> tuple[int, int]:
+    return _epoch[0], _generation.get(key, 0)
+
+
+async def _build_and_keep(app_state: Any, root: Conversation, key: str) -> RawSnapshot:
+    generation = _generation_of(key)
+    with agent_rows_memo():
         raw = await build_snapshot(app_state, root)
+    if _generation_of(key) == generation:
         _snapshots[key] = raw
-        return raw
+    return raw
+
+
+def _refresh_in_background(app_state: Any, root: Conversation, key: str) -> None:
+    """Start one rebuild of *key* unless one runs already (on this loop)."""
+    loop = asyncio.get_running_loop()
+    task_key = (id(loop), key)
+    running = _refreshes.get(task_key)
+    if running is not None and not running.done():
+        return
+
+    async def refresh() -> None:
+        lock = _locks.setdefault(task_key, asyncio.Lock())
+        try:
+            async with lock:
+                cached = _snapshots.get(key)
+                if cached is not None and time.monotonic() - cached.built_at < SNAPSHOT_TTL_S:
+                    return
+                await _build_and_keep(app_state, root, key)
+        except Exception:  # noqa: BLE001 - the next read builds it in the open
+            _logger.warning("factory: background snapshot of %s failed", key, exc_info=True)
+        finally:
+            if _refreshes.get(task_key) is task:
+                _refreshes.pop(task_key, None)
+
+    task = loop.create_task(refresh(), name=f"factory-snapshot-{key}")
+    _refreshes[task_key] = task
 
 
 def forget(factory_id: str) -> None:
     """Drop one factory's cached snapshot (a write the next read must see)."""
-    _snapshots.pop(bare_id(factory_id), None)
+    key = bare_id(factory_id)
+    _generation[key] = _generation.get(key, 0) + 1
+    _snapshots.pop(key, None)
+    with _views_lock:
+        _views.pop(key, None)
 
 
 def reset_cache() -> None:
     """Forget cached snapshots, viewer permissions, costs and handles (tests)."""
+    _epoch[0] += 1
     _snapshots.clear()
     _cost_cache.clear()
     _locks.clear()
     _viewers.clear()
+    with _views_lock:
+        _views.clear()
     identity.reset_cache()
 
 
+# ── One viewer's cut, kept with the snapshot it was cut from ───
+
+_views: dict[str, dict[frozenset[str], tuple[RawSnapshot, dict[str, Any], str]]] = {}
+_VIEWS_PER_FACTORY: Final = 32
+_views_lock = threading.Lock()
+
+
+def filtered_view(raw: RawSnapshot, readable: Iterable[str]) -> tuple[dict[str, Any], str]:
+    """
+    :func:`filter_snapshot` of *raw* for *readable*, with a digest of it,
+    computed once per snapshot and set of readable sessions.
+
+    :returns: ``(body, digest)``; the body is shared: callers copy before changing it.
+    """
+    who = frozenset(readable)
+    with _views_lock:
+        hit = _views.get(raw.factory_id, {}).get(who)
+    if hit is not None and hit[0] is raw:
+        return hit[1], hit[2]
+    body = filter_snapshot(raw, set(who))
+    digest = _stable_digest(body)
+    with _views_lock:
+        per = _views.setdefault(raw.factory_id, {})
+        if len(per) >= _VIEWS_PER_FACTORY or any(r is not raw for r, _, _ in per.values()):
+            per.clear()
+        per[who] = (raw, body, digest)
+    return body, digest
+
+
 def readable_set(
     user_id: str | None,
     session_ids: Iterable[str],
@@ -1702,16 +1821,44 @@ def readable_set(
 
     Synchronous (run it in a worker thread).
     """
-    from omnigent.server.factory.org_chart import _may_read
-
     ids = set(session_ids)
-    if deps.permission_store is None:
+    perms = deps.permission_store
+    if perms is None:
         return ids
-    return {
-        sid
-        for sid in ids
-        if _may_read(user_id, sid, deps.conversation_store, deps.permission_store)
+    if user_id is None or not ids:
+        return set()
+    # check_session_access for every id, in a few batched reads: admin bypass,
+    # a sub-agent reads as its top-level session, then a direct or public grant.
+    if perms.is_admin(user_id):
+        return ids
+    store = deps.conversation_store
+    top_of: dict[str, str] = {}
+    pending = {sid: sid for sid in ids}
+    for _ in range(_MAX_PARENT_HOPS):
+        if not pending:
+            break
+        rows = store.get_conversations(list(set(pending.values())))
+        nxt: dict[str, str] = {}
+        for sid, at in pending.items():
+            conv = rows.get(at)
+            if conv is None:
+                continue
+            if conv.parent_conversation_id is None:
+                top_of[sid] = at
+            else:
+                nxt[sid] = conv.parent_conversation_id
+        pending = nxt
+    if not top_of:
+        return set()
+    grants = perms.list_for_sessions(sorted(set(top_of.values())))
+    allowed = {
+        cid
+        for cid, rows in grants.items()
+        if any(
+            g.user_id in (user_id, RESERVED_USER_PUBLIC) and g.level >= LEVEL_READ for g in rows
+        )
     }
+    return {sid for sid, top in top_of.items() if top in allowed}
 
 
 def viewer_readable(raw: RawSnapshot, readable_top: set[str]) -> set[str]:
@@ -1832,15 +1979,24 @@ def snapshot_etag(body: Mapping[str, Any]) -> str:
     ``elapsed_s`` counters, which the cockpit ticks itself), so an unchanged
     factory answers 304.
     """
+    return f'W/"{_stable_digest(body)[:32]}"'
+
+
+def view_etag(digest: str, extras: Mapping[str, Any]) -> str:
+    """A weak ETag over a cut's digest (:func:`filtered_view`) and the viewer's own fields."""
+    tail = json.dumps(extras, sort_keys=True, separators=(",", ":"), default=str)
+    return f'W/"{hashlib.sha256((digest + tail).encode()).hexdigest()[:32]}"'
+
+
+def _stable_digest(body: Mapping[str, Any]) -> str:
     stable = {k: v for k, v in body.items() if k != "server_time"}
     stable["factory"] = {k: v for k, v in body["factory"].items() if k != "elapsed_s"}
     stable["projects"] = [
         {k: v for k, v in p.items() if k != "elapsed_s"} for p in body["projects"]
     ]
-    digest = hashlib.sha256(
+    return hashlib.sha256(
         json.dumps(stable, sort_keys=True, separators=(",", ":"), default=str).encode()
     ).hexdigest()
-    return f'W/"{digest[:32]}"'
 
 
 def summary_from(body: Mapping[str, Any]) -> dict[str, Any]:
@@ -1885,6 +2041,7 @@ def summary_from(body: Mapping[str, Any]) -> dict[str, Any]:
     "decision_view",
     "delivered_brief_targets",
     "filter_snapshot",
+    "filtered_view",
     "forget",
     "get_snapshot",
     "is_factory_root",
@@ -1903,5 +2060,6 @@ def summary_from(body: Mapping[str, Any]) -> dict[str, Any]:
     "start_view",
     "subtree_cost",
     "summary_from",
+    "view_etag",
     "viewer_readable",
 ]
diff --git a/omnigent/server/factory/tasks.py b/omnigent/server/factory/tasks.py
index 0d180c5d4e2..6418fde66e9 100644
--- a/omnigent/server/factory/tasks.py
+++ b/omnigent/server/factory/tasks.py
@@ -28,12 +28,13 @@
 import re
 import time
 from collections import defaultdict
-from collections.abc import Collection, Iterable, Mapping, Sequence
+from collections.abc import Callable, Collection, Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
 from typing import Any, Final, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
+from omnigent.entities import Conversation
 from omnigent.entities.factory import FactoryDecision, FactoryPr, FactoryProject, FactoryTask
 from omnigent.factory import (
     LABEL_PREFIX,
@@ -421,24 +422,32 @@ def _decision_met(decision: FactoryDecision | None) -> bool:
     return decision is not None and decision.status in ("answered", "withdrawn")
 
 
-def _is_paused(deps: FactoryDeps, factory_id: str, project: FactoryProject | None) -> bool:
-    store = deps.conversation_store
-    root = store.get_conversation(factory_id)
+ConvGetter = Callable[[str], "Conversation | None"]
+
+
+def _is_paused(
+    deps: FactoryDeps,
+    factory_id: str,
+    project: FactoryProject | None,
+    get: ConvGetter | None = None,
+) -> bool:
+    get = get or deps.conversation_store.get_conversation
+    root = get(factory_id)
     if root is not None and root.labels.get(LABEL_STATE) == STATE_PAUSED:
         return True
     if project is not None and project.pl_session_id:
-        pl = store.get_conversation(project.pl_session_id)
+        pl = get(project.pl_session_id)
         return pl is not None and pl.labels.get(LABEL_STATE) == STATE_PAUSED
     return False
 
 
-def _worker_running(deps: FactoryDeps, task: FactoryTask) -> bool:
+def _worker_running(deps: FactoryDeps, task: FactoryTask, get: ConvGetter | None = None) -> bool:
     """:returns: Whether one of the task's bound sessions is running now."""
     from omnigent.server.routes._sessions.helpers import _session_status_from_cache
 
-    store = deps.conversation_store
+    get = get or deps.conversation_store.get_conversation
     for session_id in task.session_ids or ():
-        conv = store.get_conversation(session_id)
+        conv = get(session_id)
         if conv is not None and _session_status_from_cache(conv.id, conv.live_status) == "running":
             return True
     return False
@@ -503,6 +512,26 @@ class FactoryData:
     others: list[FactoryTask] = field(default_factory=list)
     """Other projects' open tasks, when only one project's tasks were read (a
     ``project:<slug>`` dependency is not met while its project still works)."""
+    convs: dict[str, Conversation | None] = field(default_factory=dict, repr=False)
+    """Sessions read for this load (paused labels, bound workers), each once."""
+
+    def conv(self, deps: FactoryDeps, session_id: str) -> Conversation | None:
+        """:returns: The session, read once per load."""
+        if session_id not in self.convs:
+            self.convs[session_id] = deps.conversation_store.get_conversation(session_id)
+        return self.convs[session_id]
+
+    def prefetch(self, deps: FactoryDeps) -> None:
+        """Read the root, the PLs and the tasks' workers in one batch."""
+        ids = {self.factory_id}
+        ids |= {p.pl_session_id for p in self.projects.values() if p.pl_session_id}
+        ids |= {sid for t in self.tasks for sid in t.session_ids or ()}
+        ids -= set(self.convs)
+        if not ids:
+            return
+        found = deps.conversation_store.get_conversations(sorted(ids))
+        for sid in ids:
+            self.convs[sid] = found.get(sid)
 
     def project_ctx(self, deps: FactoryDeps, slug: str) -> ProjectCtx:
         """:returns: The context of one project (paused state read from labels)."""
@@ -523,12 +552,16 @@ def project_ctx(self, deps: FactoryDeps, slug: str) -> ProjectCtx:
                     ref = dep.removeprefix(DECISION_DEP_PREFIX)
                     external[dep] = _decision_met(self.decisions.get(ref))
         active = project is None or project.status in (None, "active")
+
+        def get(session_id: str) -> Conversation | None:
+            return self.conv(deps, session_id)
+
         # Ad-hoc work holds an IC slot only while one of its workers runs (the
         # PL never started it, so nothing else closes it).
         idle = {
             t.id
             for t in tasks
-            if t.kind == "adhoc" and t.state in OCCUPYING and not _worker_running(deps, t)
+            if t.kind == "adhoc" and t.state in OCCUPYING and not _worker_running(deps, t, get)
         }
         return ProjectCtx(
             factory_id=self.factory_id,
@@ -537,7 +570,7 @@ def project_ctx(self, deps: FactoryDeps, slug: str) -> ProjectCtx:
             tasks=tasks,
             charter=charter,
             external_met=external,
-            paused=_is_paused(deps, self.factory_id, project),
+            paused=_is_paused(deps, self.factory_id, project, get),
             active=active,
             slots=_slots(charter, tasks, self.prs, slug, idle, _team_builders(deps, self, slug)),
             observed=self.observed,
@@ -726,6 +759,7 @@ def snapshot_fields(
         return {}, {}, []
     open_tasks = [t for t in data.tasks if t.state not in TASK_CLOSED_STATES]
     try:
+        data.prefetch(deps)
         out: dict[str, dict[str, Any]] = {}
         for slug in sorted({t.project_slug for t in data.tasks}):
             ctx = data.project_ctx(deps, slug)
diff --git a/omnigent/stores/conversation_store/__init__.py b/omnigent/stores/conversation_store/__init__.py
index e4482a8e9cb..e0025e85f28 100644
--- a/omnigent/stores/conversation_store/__init__.py
+++ b/omnigent/stores/conversation_store/__init__.py
@@ -580,6 +580,27 @@ def list_child_conversation_ids_by_parent(
         """
         ...
 
+    def list_sub_agents_by_parent(
+        self, parent_conversation_ids: list[str], limit: int = 200
+    ) -> dict[str, list[Conversation]]:
+        """
+        The newest *limit* unarchived sub-agents of each parent, oldest first.
+
+        Batched counterpart to ``list_conversations(kind="sub_agent",
+        parent_conversation_id=p, order="desc", limit=limit)`` per parent
+        (reversed). This default runs exactly that; stores override it with a
+        constant number of queries.
+
+        :returns: Every unique input parent id mapped to its children.
+        """
+        out: dict[str, list[Conversation]] = {}
+        for parent_id in dict.fromkeys(parent_conversation_ids):
+            page = self.list_conversations(
+                limit=limit, kind="sub_agent", parent_conversation_id=parent_id, order="desc"
+            )
+            out[parent_id] = list(reversed(page.data))
+        return out
+
     @abstractmethod
     def list_items(
         self,
diff --git a/omnigent/stores/conversation_store/sqlalchemy_store.py b/omnigent/stores/conversation_store/sqlalchemy_store.py
index 11fd03ce9ee..c17a0478e81 100644
--- a/omnigent/stores/conversation_store/sqlalchemy_store.py
+++ b/omnigent/stores/conversation_store/sqlalchemy_store.py
@@ -1326,6 +1326,35 @@ def list_child_conversation_ids_by_parent(
                 result[parent_id].append(child_id)
         return result
 
+    def list_sub_agents_by_parent(
+        self, parent_conversation_ids: list[str], limit: int = 200
+    ) -> dict[str, list[Conversation]]:
+        """One id query plus :meth:`get_conversations`. See base class for contract."""
+        unique_ids = list(dict.fromkeys(parent_conversation_ids))
+        result: dict[str, list[Conversation]] = {parent_id: [] for parent_id in unique_ids}
+        if not unique_ids:
+            return result
+        with self._conv_session("list_sub_agents_by_parent") as ap_sess:
+            rows = ap_sess.execute(
+                select(SqlConversation.parent_conversation_id, SqlConversation.id)
+                .where(SqlConversation.workspace_id == current_workspace_id())
+                .where(SqlConversation.parent_conversation_id.in_(unique_ids))
+                .where(SqlConversation.archived.is_(False))
+                .order_by(
+                    SqlConversation.parent_conversation_id,
+                    desc(SqlConversation.created_at),
+                    desc(self._tiebreaker_col),
+                )
+            ).all()
+        newest: dict[str, list[str]] = {parent_id: [] for parent_id in unique_ids}
+        for parent_id, child_id in rows:
+            if parent_id is not None and len(newest[parent_id]) < limit:
+                newest[parent_id].append(child_id)
+        convs = self.get_conversations([cid for ids in newest.values() for cid in ids])
+        for parent_id, ids in newest.items():
+            result[parent_id] = [convs[cid] for cid in reversed(ids) if cid in convs]
+        return result
+
     def set_labels(
         self,
         conversation_id: str,
diff --git a/tests/server/factory/perf_world.py b/tests/server/factory/perf_world.py
new file mode 100644
index 00000000000..641ade976da
--- /dev/null
+++ b/tests/server/factory/perf_world.py
@@ -0,0 +1,406 @@
+"""
+(sample) A factory the size of a live one, for the cockpit's read benchmark.
+
+:func:`seed_live_factory` writes the 13:40 story (``tests/e2e_ui/factory_v3``)
+and then scales it to a live factory's size: more projects with their leads and
+sub-agents, tasks and PRs, a long decision and proposal history, thousands of
+events and chat messages, demos and spans. :func:`build_app` mounts the factory
+routes over a SQLite database the way the server does (SQL factory store, real
+grants, header identities). ``dev/benchmarks/factory_cockpit.py`` and
+``test_cockpit_perf.py`` read it.
+"""
+
+from __future__ import annotations
+
+import random
+from dataclasses import dataclass
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, Final
+
+from omnigent import factory as F
+from omnigent.db.utils import builtin_agent_id
+from omnigent.entities.factory import EventDraft
+from omnigent.server.auth import LEVEL_OWNER
+from omnigent.stores.agent_store.sqlalchemy_store import SqlAlchemyAgentStore
+from omnigent.stores.artifact_store.local import LocalArtifactStore
+from omnigent.stores.conversation_store.sqlalchemy_store import SqlAlchemyConversationStore
+from omnigent.stores.factory_store.sqlalchemy_store import SqlAlchemyFactoryStore
+from omnigent.stores.permission_store.sqlalchemy_store import SqlAlchemyPermissionStore
+from tests.server.helpers import build_agent_bundle
+
+OWNER: Final = "ana@perf.test"
+REPO: Final = "sample-org/sample-app"
+PROPOSAL_KINDS: Final = ("projects", "rules", "tuning")
+TASK_STATES: Final = ("done", "done", "running", "ready", "planned", "review", "blocked")
+
+
+@dataclass(frozen=True)
+class Scale:
+    """How big the seeded factory is (the defaults are a live factory's size)."""
+
+    projects: int = 20
+    subagents_per_pl: int = 3
+    tasks_per_project: int = 8
+    prs_per_project: int = 4
+    events: int = 5000
+    decisions: int = 300
+    proposals: int = 300
+    demos: int = 30
+    chat: int = 2000
+    spans: int = 600
+
+
+SMALL: Final = Scale(
+    projects=12,
+    subagents_per_pl=2,
+    tasks_per_project=5,
+    prs_per_project=2,
+    events=1500,
+    decisions=120,
+    proposals=120,
+    demos=12,
+    chat=600,
+    spans=200,
+)
+"""A smaller factory for the test suite's perf budget."""
+
+
+class EmailAuth:
+    """Header identity, as the SSO proxy hands it (``X-Forwarded-Email``)."""
+
+    def get_user_id(self, request: Any) -> str | None:
+        return request.headers.get("X-Forwarded-Email")
+
+
+def _native(role: str) -> dict[str, Any]:
+    return {"type": "omnigent", "config": {"harness": "claude-native", "factory_role": role}}
+
+
+def build_app(db_uri: str, tmp: Path) -> Any:
+    """
+    The factory routes over *db_uri* as the server mounts them (plus ``/ping``,
+    a request that does nothing, to see whether the event loop is free).
+
+    The ``hello_world`` agent the 13:40 seed runs is registered.
+    """
+    from fastapi import FastAPI
+    from fastapi.responses import JSONResponse
+
+    from omnigent.errors import OmnigentError
+    from omnigent.runtime.agent_cache import AgentCache
+    from omnigent.server.factory import register_factory_routes
+
+    conv = SqlAlchemyConversationStore(db_uri)
+    agents = SqlAlchemyAgentStore(db_uri)
+    perms = SqlAlchemyPermissionStore(db_uri)
+    artifacts = LocalArtifactStore(str(tmp / "artifacts"))
+    cache = AgentCache(artifacts, tmp / "agent-cache")
+    if agents.get_by_name("hello_world") is None:
+        agent_id = builtin_agent_id("hello_world")
+        bundle = build_agent_bundle(
+            "hello_world",
+            executor=_native("project_lead"),
+            include_llm=False,
+            sub_agents=[
+                {"name": "ic", "description": "implementer", "executor": _native("ic")},
+                {"name": "reviewer", "description": "reviewer", "executor": _native("reviewer")},
+            ],
+        )
+        artifacts.put(f"{agent_id}/v1", bundle)
+        agents.create(agent_id, "hello_world", f"{agent_id}/v1")
+    app = FastAPI()
+    register_factory_routes(
+        app,
+        conversation_store=conv,
+        agent_store=agents,
+        agent_cache=cache,
+        permission_store=perms,
+        auth_provider=EmailAuth(),
+        factory_store=SqlAlchemyFactoryStore(db_uri),
+    )
+
+    async def _omnigent_error(_request: Any, exc: Exception) -> JSONResponse:
+        assert isinstance(exc, OmnigentError)
+        return JSONResponse(
+            status_code=exc.http_status,
+            content={"error": {"code": exc.code, "message": exc.message}},
+        )
+
+    app.add_exception_handler(OmnigentError, _omnigent_error)
+
+    @app.get("/ping")
+    async def ping() -> dict[str, bool]:
+        return {"ok": True}
+
+    return app
+
+
+def seed_live_factory(db_uri: str, app: Any, scale: Scale = Scale(), *, seed: int = 7) -> str:
+    """
+    Write the 13:40 factory, then grow it to *scale*. :returns: The factory id.
+
+    The 13:40 seed files one proposal through the route; it goes to *app* in process.
+    """
+    import time
+
+    from fastapi.testclient import TestClient
+
+    from tests.e2e_ui.factory_v3 import seed as story
+
+    client = TestClient(app)
+    real_httpx = story.httpx
+
+    def _post(url: str, **kwargs: Any) -> Any:
+        path = url.split("://", 1)[-1].split("/", 1)[1]
+        kwargs.pop("timeout", None)
+        return client.post(f"/{path}", **kwargs)
+
+    story.httpx = SimpleNamespace(post=_post)  # type: ignore[assignment]
+    try:
+        seeded = story.seed_factory_1340(db_uri, "http://bench", OWNER)
+    finally:
+        story.httpx = real_httpx  # type: ignore[assignment]
+    _grow(db_uri, seeded.fid, scale, random.Random(seed), time.time())
+    return seeded.fid
+
+
+def _grow(db_uri: str, fid: str, scale: Scale, rnd: random.Random, clock: float) -> None:
+    conv = SqlAlchemyConversationStore(db_uri)
+    perms = SqlAlchemyPermissionStore(db_uri)
+    fs = SqlAlchemyFactoryStore(db_uri)
+    agent = SqlAlchemyAgentStore(db_uri).get_by_name("hello_world")
+    assert agent is not None
+    root = conv.get_conversation(fid)
+    assert root is not None
+    sentinel = root.labels.get(F.LABEL_SUPERVISOR) or fid
+    member = {F.LABEL_ID: fid, F.LABEL_PARENT: fid, F.LABEL_STATE: "active"}
+    for n in range(4):
+        fs.upsert_person(fid, f"person{n}@perf.test", role="operator", invited_by=OWNER)
+        perms.ensure_user(f"person{n}@perf.test")
+
+    existing = {p.slug for p in fs.list_projects(fid)}
+    extra = max(0, scale.projects - len(existing))
+    sessions: list[str] = [fid, sentinel]
+    leads: dict[str, str] = {}
+    for i in range(extra):
+        slug = f"perf-project-{i:02d}"
+        pl = conv.create_conversation(kind="default", title=f"PL · {slug}", agent_id=agent.id)
+        conv.set_labels(
+            pl.id,
+            {
+                **member,
+                F.LABEL_ROLE: F.ROLE_PROJECT_LEAD,
+                F.LABEL_PROJECT: slug,
+                F.LABEL_SUPERVISOR: sentinel,
+                F.LABEL_HEARTBEAT_S: "600",
+            },
+        )
+        perms.grant(OWNER, pl.id, LEVEL_OWNER)
+        conv.set_session_usage(pl.id, {"total_cost_usd": round(rnd.uniform(1, 60), 2)})
+        leads[slug] = pl.id
+        sessions.append(pl.id)
+        for k in range(scale.subagents_per_pl):
+            name = ("ic", "reviewer")[k % 2]
+            child = conv.create_conversation(
+                kind="sub_agent",
+                title=f"{name}:task-{k}",
+                parent_conversation_id=pl.id,
+                agent_id=agent.id,
+                sub_agent_name=name,
+            )
+            sessions.append(child.id)
+        stage = rnd.choice(["building", "planning", "ready", "reviewing", "merged"])
+        fs.upsert_project(
+            fid,
+            slug,
+            title=f"Perf project {i}",
+            repo=REPO,
+            base_branch="sample-main",
+            created_by="server",
+            stage=stage,
+            priority=i % 3,
+            pl_session_id=pl.id,
+            feature="I-1" if i % 4 == 0 else None,
+            charter={
+                "outcome": f"Outcome of perf project {i}",
+                "goal": "a goal",
+                "dod": ["it works", "it is tested"],
+                "depends_on": [f"perf-project-{i - 1:02d}"] if i % 5 == 4 else [],
+                "budgets": {"project_usd": 150},
+            },
+        )
+
+    pr_number = 20000
+    for slug, pl in leads.items():
+        keys = [f"step-{k}" for k in range(scale.tasks_per_project)]
+        task_ids: list[str] = []
+        for k, key in enumerate(keys):
+            state = rnd.choice(TASK_STATES)
+            task, _ = fs.upsert_task(
+                fid,
+                slug,
+                key,
+                title=f"{slug} {key}",
+                state=state,
+                deps=[keys[k - 1]] if k else [],
+                estimate_min=rnd.choice([30, 60, 120]),
+                session_ids=[pl],
+                cost_usd=round(rnd.uniform(0, 5), 2),
+            )
+            task_ids.append(task.id)
+        for k in range(scale.prs_per_project):
+            pr_number += 1
+            state = rnd.choice(["open", "merged", "merged", "closed"])
+            pr, _ = fs.upsert_pr(
+                fid,
+                REPO,
+                pr_number,
+                project_slug=slug,
+                title=f"{slug}: part {k}",
+                head_sha=f"{pr_number:07x}" + "a" * 33,
+                head_ref=f"factory/{slug}/{k}",
+                base_ref="sample-main",
+                state=state,
+                ready=state == "open" and k % 2 == 0,
+                task_id=task_ids[k % len(task_ids)],
+                gates={
+                    "draft": "pass",
+                    "checks": rnd.choice(["pass", "pending", "fail"]),
+                    "review": "pass",
+                    "mergeable": "pass",
+                    "reasons": [],
+                    "evidence": {"checks": {"passing": 12, "total": 14}},
+                },
+                **({"merged_at": int(clock - rnd.uniform(0, 86400))} if state == "merged" else {}),
+            )
+            fs.update_task(task_ids[k % len(task_ids)], pr_id=pr.id)
+
+    slugs = [*existing, *leads]
+    for n in range(scale.proposals):
+        status = rnd.choice(["rejected", "applied", "applied", "withdrawn"])
+        fs.create_proposal(
+            fid,
+            kind="projects"
+            if status == "applied"
+            else rnd.choice(["projects", "rules", "tuning"]),
+            title=f"History proposal {n} (sample)",
+            author_kind="agent",
+            author=rnd.choice(sessions),
+            status=status,
+            ops=[{"op": "update", "slug": rnd.choice(slugs), "patch": {"constraints": "x"}}]
+            if status != "applied"
+            else None,
+            result={"created": []} if status == "applied" else None,
+        )
+    for n in range(scale.decisions):
+        row, _ = fs.create_decision(
+            fid,
+            dedupe_key=f"history-{n}",
+            question=f"History question {n} (sample)?",
+            source=rnd.choice(["agent", "loop"]),
+            project_slug=rnd.choice(slugs),
+            asked_by_session_id=rnd.choice(sessions),
+        )
+        if n % 3:
+            fs.answer_decision(row.id, option_id=None, text="done", by=OWNER)
+        else:
+            fs.set_decision_status(row.id, "expired")
+    for n in range(8):
+        fs.create_decision(
+            fid,
+            dedupe_key=f"open-{n}",
+            question=f"Open question {n} (sample)?",
+            source="agent",
+            project_slug=rnd.choice(list(leads) or slugs),
+            asked_by_session_id=rnd.choice(sessions),
+        )
+
+    kinds = [
+        "message.queued",
+        "message.delivered",
+        "task.state",
+        "task.claimed",
+        "pr.updated",
+        "pr.gates",
+        "heartbeat.sent",
+        "member.state",
+        "chat.posted",
+        "presence.seen",
+        "loop.tick",
+        "decision.opened",
+    ]
+    for n in range(scale.events):
+        kind = kinds[n % len(kinds)]
+        ts = int((clock - (scale.events - n) * 17) * 1000)
+        data: dict[str, Any] = {"n": n}
+        if kind.startswith("message."):
+            data["mail_id"] = f"mail-{n // 2}"
+        fs.append_event(
+            EventDraft(
+                factory_id=fid,
+                kind=kind,
+                summary=f"{kind} #{n} (sample)",
+                project_slug=rnd.choice(slugs),
+                from_id=rnd.choice(sessions),
+                to_id=rnd.choice(sessions),
+                subject=f"task:T-{rnd.randint(1, 100)}",
+                data=data,
+                ts=ts,
+            )
+        )
+
+    all_prs = fs.list_prs(fid)
+    for n in range(scale.demos):
+        slug = rnd.choice(list(leads) or slugs)
+        prs = [p for p in all_prs if p.project_slug == slug]
+        pr = prs[n % len(prs)] if prs else None
+        fs.create_demo(
+            factory_id=fid,
+            project_slug=slug,
+            repo=REPO,
+            pr_number=pr.number if pr else 0,
+            head_sha=pr.head_sha if pr and pr.head_sha else "abc1234",
+            session_id=leads.get(slug, fid),
+            created_by_session=leads.get(slug, fid),
+            title=f"Demo {n}",
+            summary="It renders (sample).",
+            files=[
+                {"file_id": f"f-{n}-v", "name": "demo.webm", "kind": "video", "bytes": 4_000_000},
+                {"file_id": f"f-{n}-s", "name": "shot.png", "kind": "image", "bytes": 200_000},
+            ],
+            manifest={"steps": [{"n": 1, "text": "open the app", "status": "pass"}]},
+            pr_id=pr.id if pr else None,
+        )
+
+    threads = [F.THREAD_FACTORY, *(f"{F.THREAD_PROJECT_PREFIX}{s}" for s in slugs[:10])]
+    for n in range(scale.chat):
+        human = n % 4 == 0
+        fs.append_chat(
+            fid,
+            thread=rnd.choice(threads),
+            author_kind="human" if human else "agent",
+            author=OWNER if human else rnd.choice(sessions),
+            author_role="owner" if human else None,
+            body=f"Chat message {n} about the work (sample).",
+            created_at=int(clock - (scale.chat - n) * 40),
+        )
+
+    for n in range(scale.spans):
+        start = int((clock - rnd.uniform(600, 86400)) * 1000)
+        slug = rnd.choice(list(leads) or slugs)
+        fs.insert_span(
+            fid,
+            kind="turn",
+            name="working",
+            trace_key=f"state:{n % 40}",
+            started_at=start,
+            ended_at=start + rnd.randint(60_000, 1_800_000),
+            outcome="ok",
+            project_slug=slug,
+            session_id=leads.get(slug, fid),
+            attrs={"segment": "working", "cost_start": 1.0, "cost_end": 1.5},
+        )
+
+
+__all__ = ["OWNER", "SMALL", "EmailAuth", "Scale", "build_app", "seed_live_factory"]
diff --git a/tests/server/factory/test_cockpit_perf.py b/tests/server/factory/test_cockpit_perf.py
new file mode 100644
index 00000000000..acbe601f8c7
--- /dev/null
+++ b/tests/server/factory/test_cockpit_perf.py
@@ -0,0 +1,130 @@
+"""
+The cockpit's reads on a factory shaped like a live one stay cheap.
+
+Seeds :data:`perf_world.SMALL` (``dev/benchmarks/factory_cockpit.py`` runs the
+live size) and reads each cockpit page cold (every cache dropped) and warm.
+SQL statements are the budget that does not flake: on Postgres each one is a
+network round trip, and an N+1 read grows them with every member and project.
+The time budgets are loose, for a slow CI box.
+"""
+
+from __future__ import annotations
+
+import time
+from collections.abc import AsyncIterator, Iterator
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pytest
+import pytest_asyncio
+from sqlalchemy import event
+
+from omnigent.db.utils import get_or_create_engine
+from omnigent.server.factory import nodes, snapshot, status
+from tests.server.factory import perf_world
+
+PAGES: dict[str, str] = {
+    "snapshot": "/v1/factory/factories/{fid}",
+    "node_product": "/v1/factory/factories/{fid}/n/product",
+    "needs_you": "/v1/factory/factories/{fid}/needs-you",
+    "chat": "/v1/factory/factories/{fid}/chat?thread=factory",
+    "status_summary": "/v1/factory/factories/{fid}/status-summary",
+    "events": "/v1/factory/factories/{fid}/events",
+    "factories": "/v1/factory/factories",
+    "decisions": "/v1/factory/decisions",
+    "preflight": "/v1/factory/preflight",
+}
+
+COLD_QUERIES: dict[str, int] = {
+    "snapshot": 105,
+    "node_product": 125,
+    "needs_you": 115,
+    "chat": 105,
+    "status_summary": 105,
+    "events": 100,
+    "factories": 105,
+    "decisions": 35,
+    "preflight": 5,
+}
+"""SQL statements a cold read may take (after the fixes: about half of each)."""
+
+WARM_P95_S = 0.3
+COLD_MAX_S = 3.0
+
+
+def _reset() -> None:
+    snapshot.reset_cache()
+    nodes.reset_cache()
+    status.reset()
+
+
+@pytest.fixture(autouse=True)
+def _fresh_caches() -> Iterator[None]:
+    _reset()
+    yield
+    _reset()
+
+
+class _Counter:
+    def __init__(self) -> None:
+        self.count = 0
+
+    def __call__(self, _conn: Any, _cursor: Any, statement: str, *_args: Any) -> None:
+        if not statement.startswith("PRAGMA"):  # SQLite's per-connection setup
+            self.count += 1
+
+
+@pytest_asyncio.fixture()
+async def live_like(db_uri: str, tmp_path: Path) -> AsyncIterator[tuple[httpx.AsyncClient, str]]:
+    app = perf_world.build_app(db_uri, tmp_path)
+    fid = perf_world.seed_live_factory(db_uri, app, perf_world.SMALL)
+    transport = httpx.ASGITransport(app=app)
+    headers = {"X-Forwarded-Email": perf_world.OWNER}
+    async with httpx.AsyncClient(transport=transport, base_url="http://t", headers=headers) as c:
+        yield c, fid
+
+
+async def test_cockpit_reads_stay_within_their_query_and_time_budgets(
+    live_like: tuple[httpx.AsyncClient, str], db_uri: str
+) -> None:
+    client, fid = live_like
+    counter = _Counter()
+    engine = get_or_create_engine(db_uri)
+    event.listen(engine, "before_cursor_execute", counter)
+    try:
+        # Process-wide warm-ups (definition seed, agent bundles) are not a page's cost.
+        for path in PAGES.values():
+            assert (await client.get(path.format(fid=fid))).status_code == 200
+        over: list[str] = []
+        for name, path in PAGES.items():
+            url = path.format(fid=fid)
+            _reset()
+            before, started = counter.count, time.perf_counter()
+            resp = await client.get(url)
+            cold_s, queries = time.perf_counter() - started, counter.count - before
+            assert resp.status_code == 200, (name, resp.text[:300])
+            warm = []
+            for _ in range(8):
+                started = time.perf_counter()
+                assert (await client.get(url)).status_code == 200
+                warm.append(time.perf_counter() - started)
+            p95 = sorted(warm)[-1]
+            if queries > COLD_QUERIES[name]:
+                over.append(f"{name}: {queries} SQL statements cold (budget {COLD_QUERIES[name]})")
+            if cold_s > COLD_MAX_S:
+                over.append(f"{name}: {cold_s:.2f}s cold (budget {COLD_MAX_S}s)")
+            if p95 > WARM_P95_S:
+                over.append(f"{name}: {p95 * 1000:.0f} ms warm p95 (budget {WARM_P95_S}s)")
+        assert not over, "\n".join(over)
+    finally:
+        event.remove(engine, "before_cursor_execute", counter)
+
+
+async def test_an_unchanged_snapshot_answers_304(live_like: tuple[httpx.AsyncClient, str]) -> None:
+    client, fid = live_like
+    first = await client.get(f"/v1/factory/factories/{fid}")
+    again = await client.get(
+        f"/v1/factory/factories/{fid}", headers={"If-None-Match": first.headers["etag"]}
+    )
+    assert again.status_code == 304
diff --git a/web/src/hooks/SessionUpdatesProvider.test.tsx b/web/src/hooks/SessionUpdatesProvider.test.tsx
index d4cf92b6def..d287474cbf0 100644
--- a/web/src/hooks/SessionUpdatesProvider.test.tsx
+++ b/web/src/hooks/SessionUpdatesProvider.test.tsx
@@ -33,6 +33,7 @@ vi.mock("@/lib/sessionUpdatesSocket", () => ({
 }));
 
 import { SessionUpdatesProvider } from "./SessionUpdatesProvider";
+import { FACTORY_FRAME_MIN_MS } from "./useFactory";
 
 function conv(id: string): Conversation {
   return {
@@ -433,6 +434,37 @@ describe("SessionUpdatesProvider projects_changed frames", () => {
 });
 
 describe("SessionUpdatesProvider factory_changed frames", () => {
+  beforeEach(() => {
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+  // Frames further apart than the throttle window each refresh at once.
+  const later = () => act(() => vi.advanceTimersByTime(FACTORY_FRAME_MIN_MS));
+
+  it("coalesces a burst of frames into one more refresh at the end of the window", () => {
+    const client = new QueryClient();
+    seedConversations(client, ["conv_a"]);
+    renderProvider(client, ["/"]);
+    const invalidate = vi.spyOn(client, "invalidateQueries");
+    const keys = () => invalidate.mock.calls.map(([filters]) => filters?.queryKey);
+    const frame = (scopes: string[]) =>
+      act(() => frameHandler()({ type: "factory_changed", factory_id: "f4f4ba9c", scopes }));
+
+    frame(["now"]);
+    expect(keys()).toEqual([["factory", "f4f4ba9c"]]);
+    invalidate.mockClear();
+    frame(["now"]);
+    frame(["presence"]);
+    frame(["snapshot", "events", "definition"]);
+    expect(keys()).toEqual([]);
+    later();
+    expect(keys()).toContainEqual(["factory-events", "f4f4ba9c"]);
+    expect(keys()).toContainEqual(["factory-team", "f4f4ba9c"]);
+    expect(keys().filter((k) => k[0] === "factory")).toHaveLength(1);
+  });
+
   it("invalidates the factory snapshot, its events, the factory list and decisions", () => {
     const client = new QueryClient();
     seedConversations(client, ["conv_a"]);
@@ -475,6 +507,7 @@ describe("SessionUpdatesProvider factory_changed frames", () => {
       }),
     );
     expect(keys()).not.toContainEqual(["factory-team", "f4f4ba9c"]);
+    later();
 
     act(() =>
       frameHandler()({
@@ -496,9 +529,11 @@ describe("SessionUpdatesProvider factory_changed frames", () => {
       act(() => frameHandler()({ type: "factory_changed", factory_id: "f4f4ba9c", scopes }));
 
     frame(["snapshot", "events", "tasks"]);
+    later();
     frame(["snapshot", "events", "chat"]);
     expect(keys()).not.toContainEqual(["factory-discovery", "f4f4ba9c"]);
     for (const scope of ["discovery", "proposals", "definition"]) {
+      later();
       invalidate.mockClear();
       frame(["snapshot", "events", scope]);
       expect(keys()).toContainEqual(["factory-discovery", "f4f4ba9c"]);
@@ -512,6 +547,7 @@ describe("SessionUpdatesProvider factory_changed frames", () => {
     const invalidate = vi.spyOn(client, "invalidateQueries");
 
     act(() => frameHandler()({ type: "factory_changed", factory_id: "f4f4ba9c", scopes: ["now"] }));
+    later();
     act(() =>
       frameHandler()({ type: "factory_changed", factory_id: "f4f4ba9c", scopes: ["presence"] }),
     );
diff --git a/web/src/hooks/SessionUpdatesProvider.tsx b/web/src/hooks/SessionUpdatesProvider.tsx
index 66af2812983..06ff47f0d52 100644
--- a/web/src/hooks/SessionUpdatesProvider.tsx
+++ b/web/src/hooks/SessionUpdatesProvider.tsx
@@ -20,7 +20,7 @@ import { type QueryClient, useQueryClient } from "@tanstack/react-query";
 import { getCurrentUserId } from "@/lib/identity";
 import { useActiveConversationId } from "@/hooks/useActiveConversationId";
 import { childSessionsQueryKey, type ChildSessionInfo } from "@/hooks/useChildSessions";
-import { factoryInvalidationKeys } from "@/hooks/useFactory";
+import { createFactoryFrameThrottle, factoryInvalidationKeys } from "@/hooks/useFactory";
 import { FACTORY_PARENT_LABEL_KEY } from "@/lib/factory";
 import {
   isSessionArchiving,
@@ -317,6 +317,12 @@ export function SessionUpdatesProvider({ children }: { children: ReactNode }) {
       }
     };
 
+    // Busy factories send frames every half second: refresh each at most every 2 s.
+    const factoryFrames = createFactoryFrameThrottle((factoryId, scopes) => {
+      for (const queryKey of factoryInvalidationKeys(factoryId, scopes)) {
+        void queryClient.invalidateQueries({ queryKey });
+      }
+    });
     const unsubscribeFrames = sessionUpdatesSocket.subscribe((frame: SessionUpdatesFrame) => {
       switch (frame.type) {
         case "heartbeat":
@@ -335,9 +341,7 @@ export function SessionUpdatesProvider({ children }: { children: ReactNode }) {
         case "factory_changed":
           // A factory's board, activity or decisions moved; the snapshot
           // refetch is ETag-conditional, so a no-op change costs a 304.
-          for (const queryKey of factoryInvalidationKeys(frame.factory_id, frame.scopes)) {
-            void queryClient.invalidateQueries({ queryKey });
-          }
+          factoryFrames.push(frame.factory_id, frame.scopes);
           return;
         case "removed":
           for (const id of frame.ids) commentsFingerprintsRef.current.delete(id);
@@ -424,6 +428,7 @@ export function SessionUpdatesProvider({ children }: { children: ReactNode }) {
 
     return () => {
       unsubscribeFrames();
+      factoryFrames.dispose();
       unsubscribeCache();
       if (invalidateTimer !== null) clearTimeout(invalidateTimer);
       if (watchTimer !== null) clearTimeout(watchTimer);
diff --git a/web/src/hooks/useFactory.ts b/web/src/hooks/useFactory.ts
index d6ba8390d92..09c1b89fb30 100644
--- a/web/src/hooks/useFactory.ts
+++ b/web/src/hooks/useFactory.ts
@@ -135,6 +135,72 @@ export function factoryInvalidationKeys(
   ];
 }
 
+/**
+ * At most one refresh per factory this often while frames keep coming: the
+ * first frame applies at once, later ones merge their scopes into one refresh
+ * at the end of the window (agents record several events a second).
+ */
+export const FACTORY_FRAME_MIN_MS = 2_000;
+
+export interface FactoryFrameThrottle {
+  push(factoryId: string, scopes: readonly string[] | null | undefined): void;
+  dispose(): void;
+}
+
+/**
+ * Coalesce `factory_changed` frames per factory (leading and trailing edge).
+ *
+ * @param apply - Refreshes one factory; `null` scopes mean everything.
+ */
+export function createFactoryFrameThrottle(
+  apply: (factoryId: string, scopes: readonly string[] | null) => void,
+  minMs: number = FACTORY_FRAME_MIN_MS,
+): FactoryFrameThrottle {
+  interface Slot {
+    last: number;
+    timer: ReturnType<typeof setTimeout> | null;
+    // `undefined`: nothing waiting; `null`: a frame without scopes (everything).
+    waiting: Set<string> | null | undefined;
+  }
+  const slots = new Map<string, Slot>();
+  const flush = (factoryId: string, slot: Slot) => {
+    slot.timer = null;
+    const waiting = slot.waiting;
+    slot.waiting = undefined;
+    if (waiting === undefined) return;
+    slot.last = Date.now();
+    apply(factoryId, waiting === null ? null : [...waiting]);
+  };
+  return {
+    push(factoryId, scopes) {
+      const now = Date.now();
+      let slot = slots.get(factoryId);
+      if (!slot) {
+        slot = { last: -Infinity, timer: null, waiting: undefined };
+        slots.set(factoryId, slot);
+      }
+      if (slot.timer === null && now - slot.last >= minMs) {
+        slot.last = now;
+        apply(factoryId, scopes?.length ? [...scopes] : null);
+        return;
+      }
+      if (!scopes?.length || slot.waiting === null) slot.waiting = null;
+      else slot.waiting = new Set([...(slot.waiting ?? []), ...scopes]);
+      if (slot.timer === null) {
+        const current = slot;
+        current.timer = setTimeout(
+          () => flush(factoryId, current),
+          Math.max(0, minMs - (now - slot.last)),
+        );
+      }
+    },
+    dispose() {
+      for (const slot of slots.values()) if (slot.timer !== null) clearTimeout(slot.timer);
+      slots.clear();
+    },
+  };
+}
+
 // One quick retry for a blip; a 4xx (no access, no such factory, routes not
 // deployed) won't heal, and the fallback poll or next frame retries anyway.
 function retryTransient(failureCount: number, error: unknown): boolean {

From f3877b45ab3541a6e2cad29af4d6840c2c0618ca Mon Sep 17 00:00:00 2001
From: Luka Skugor <luka.skugor@dfinity.org>
Date: Fri, 6 Nov 2026 10:34:20 +0100
Subject: [PATCH 2/2] factory: forget pending background rebuilds with the
 caches

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 omnigent/server/factory/nodes.py    | 1 +
 omnigent/server/factory/snapshot.py | 1 +
 2 files changed, 2 insertions(+)

diff --git a/omnigent/server/factory/nodes.py b/omnigent/server/factory/nodes.py
index 62573af811d..804ae800396 100644
--- a/omnigent/server/factory/nodes.py
+++ b/omnigent/server/factory/nodes.py
@@ -3767,6 +3767,7 @@ def reset_cache() -> None:
     """Forget every projection (tests)."""
     _epoch[0] += 1
     _trees.clear()
+    _refreshes.clear()
     _locks.clear()
 
 
diff --git a/omnigent/server/factory/snapshot.py b/omnigent/server/factory/snapshot.py
index 518280825b8..e61cd8e9574 100644
--- a/omnigent/server/factory/snapshot.py
+++ b/omnigent/server/factory/snapshot.py
@@ -1774,6 +1774,7 @@ def reset_cache() -> None:
     """Forget cached snapshots, viewer permissions, costs and handles (tests)."""
     _epoch[0] += 1
     _snapshots.clear()
+    _refreshes.clear()
     _cost_cache.clear()
     _locks.clear()
     _viewers.clear()