fix(sessions): refill starved projects when projection under-delivers

The GROUP BY probe that finds the projects a global assigned query starved
only fired when that query came back logically FULL. A query can instead
stay short because its bounded raw candidate window was entirely consumed
by compression segments and hidden rows, while older assigned
conversations still exist below it. Read as "the database has nothing more
to give", that shortfall skipped the probe and left the quiet projects
unreachable.

Have read_importable_agent_session_rows report whether the final raw
candidate window was exhausted while its logical projection stayed short
(opt-in via return_window_exhaustion, so the list API is unchanged for
every other caller), and gate the probe on saturation OR that exhaustion.
A short result from an unfilled raw window has genuinely seen every
candidate, so the common one-global-query fast path still pays no probe.

api/agent_sessions.py

@@ -685,7 +685,8 @@ def read_importable_agent_session_rows(
     include_sources: tuple[str, ...] | None = None,
     project_assignment: str | None = None,
     project_ids: tuple[str, ...] | None = None,
-) -> list[dict]:
+    return_window_exhaustion: bool = False,
+) -> list[dict] | tuple[list[dict], bool]:
     """Return agent sessions projected as importable conversations.
 
     Hermes Agent can create rows in ``state.db.sessions`` before a session has
@@ -728,6 +729,13 @@ def read_importable_agent_session_rows(
     result can exceed ``limit`` by the number of such anchors. Callers must
     therefore iterate the result rather than assume ``len(rows) <= limit``.
 
+    ``return_window_exhaustion=True`` returns ``(rows, exhausted)`` instead of
+    just ``rows``. ``exhausted`` is true only when the final bounded raw-candidate
+    window was full while its logical projection still under-delivered ``limit``.
+    That lets a caller distinguish a genuinely short database from a compressed
+    or filtered projection that needs a narrow recovery query, without paying a
+    count probe for ordinary short windows.
+
     That recovery is deliberately bounded by the oversampled candidate set
     (the ``CANDIDATE_WINDOW_MULTIPLIERS`` window that actually ran — ``limit *
     8``, or ``limit * 32`` when the first window was consumed and re-widened):
@@ -736,6 +744,10 @@ def read_importable_agent_session_rows(
     children render top-level, exactly as they did before. Widening that window
     is a ``CANDIDATE_WINDOW_MULTIPLIERS`` change, not a change to this walk.
     """
+    def _result(rows: list[dict], exhausted: bool = False):
+        """Preserve the list API unless this caller needs raw-window status."""
+        return (rows, exhausted) if return_window_exhaustion else rows
+
     wanted_project_ids: tuple[str, ...] = ()
     if project_ids is not None:
         if str(project_assignment or '').strip().lower() != 'assigned':
@@ -748,10 +760,10 @@ def read_importable_agent_session_rows(
         if not wanted_project_ids:
             # An empty narrowing selects nothing. Returning [] keeps that exact
             # instead of degrading to "every assigned conversation".
-            return []
+            return _result([])
     db_path = Path(db_path)
     if not db_path.exists():
-        return []
+        return _result([])
 
     log = log or logger
     conn = open_state_db_readonly(db_path, log=log)
@@ -775,7 +787,7 @@ def read_importable_agent_session_rows(
                     "Upgrade hermes-agent to fix this.",
                     db_path,
                 )
-            return []
+            return _result([])
 
         parent_expr = _optional_col('parent_session_id', session_cols)
         session_source_expr = _optional_col('session_source', session_cols)
@@ -865,7 +877,7 @@ def read_importable_agent_session_rows(
                 # and every row it does have is unassigned. Keep both answers
                 # exact instead of emitting SQL against a missing column.
                 if wanted == 'assigned':
-                    return []
+                    return _result([])
             elif {'parent_session_id', 'end_reason'} <= session_cols:
                 continuation_checks = [
                     "parent.end_reason IN ('compression', 'cli_close')",
@@ -1006,7 +1018,7 @@ def read_importable_agent_session_rows(
         if limit is not None:
             result_limit = max(0, int(limit))
             if result_limit == 0:
-                return []
+                return _result([])
             # The sidebar only needs a small visible window. Bound the expensive
             # messages join to a recent-activity candidate set instead of
             # aggregating every historical Hermes state.db session before
@@ -1060,14 +1072,20 @@ def read_importable_agent_session_rows(
             # fully consumed and still came up short instead of silently
             # truncating the caller's request.
             projected: list[dict] = []
+            window_exhausted = False
             for multiplier in CANDIDATE_WINDOW_MULTIPLIERS:
                 candidate_limit = max(result_limit * multiplier, result_limit)
                 cur.execute(candidate_sql, [*params, candidate_limit])
                 raw_rows = [dict(row) for row in cur.fetchall()]
                 projected = _project(raw_rows)
-                if len(projected) >= result_limit or len(raw_rows) < candidate_limit:
-                    # Either the request is satisfied or the window was not the
-                    # binding constraint (the db has nothing more to give).
+                window_exhausted = (
+                    len(raw_rows) >= candidate_limit
+                    and len(projected) < result_limit
+                )
+                if not window_exhausted:
+                    # Either the request is satisfied or the raw candidate window
+                    # was not binding, so projection has seen every candidate this
+                    # query can reveal.
                     break
             selected = projected[:result_limit]
 
@@ -1107,7 +1125,7 @@ def read_importable_agent_session_rows(
                 selected.append(parent)
                 have.add(parent_id)
                 pending.append(parent)
-            return selected
+            return _result(selected, window_exhausted)
 
         cur.execute(
             f"""
@@ -1120,7 +1138,7 @@ def read_importable_agent_session_rows(
             """,
             params,
         )
-        return _project([dict(row) for row in cur.fetchall()])
+        return _result(_project([dict(row) for row in cur.fetchall()]))
 
 
 def read_assigned_project_row_counts(

api/models.py

@@ -16,7 +16,7 @@ import uuid
 from contextlib import closing, contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Literal, overload
+from typing import Literal, cast, overload
 
 try:  # pragma: no cover - platform-specific imports.
     import fcntl as _fcntl
@@ -74,7 +74,8 @@ PROJECT_ASSIGNED_CLI_SCAN_CEILING = 2000
 # count: any fixed cap would recreate the starvation it repairs — every starved
 # project past the cap stays unreachable on every rebuild (greptile P1 on
 # #6659). Its work is bounded by construction instead: it fires only when the
-# global assigned window came back saturated; each scoped query is limited to
+# global assigned query is logically saturated OR its final raw candidate window
+# was exhausted while projection stayed short; each scoped query is limited to
 # one project's remaining budget; and the sum of those budgets across every
 # starved project is at most effective_limit * len(projects), which the scan
 # ceiling above already caps. Worst case per build: 1 global query + 1 GROUP BY
@@ -8620,22 +8621,34 @@ def _load_cli_sessions_uncached(
                 # One global newest-first query covers every profile whose whole
                 # assigned history fits in the window — the overwhelmingly common
                 # case, and the only query most builds pay.
-                global_rows = read_importable_agent_session_rows(
-                    db_path,
-                    limit=query_limit,
-                    log=logger,
-                    exclude_sources=interactive_excluded,
-                    project_assignment='assigned',
+                global_rows, global_window_exhausted = cast(
+                    tuple[list[dict], bool],
+                    read_importable_agent_session_rows(
+                        db_path,
+                        limit=query_limit,
+                        log=logger,
+                        exclude_sources=interactive_excluded,
+                        project_assignment='assigned',
+                        return_window_exhaustion=True,
+                    ),
                 )
                 _spend_assigned_rows(global_rows)
 
-                # When that window came back FULL it truncated by recency, so a
-                # single busy project can have consumed all of it and left
-                # quieter projects with nothing — their chips would show an empty
-                # project even though the rows exist (greptile P1 on #6659). A
-                # short window cannot hide anything: it already saw every
-                # assigned conversation the db has, so no follow-up is paid.
-                if effective_limit and query_limit and len(global_rows) >= query_limit:
+                # A full logical result truncates by recency, as does a result
+                # that stays short only because the final raw candidate window
+                # was consumed by compression/visibility projection.  In either
+                # case older rows can still exist, so only then pay the GROUP BY
+                # probe that identifies projects needing a scoped top-up. A short
+                # result from an unfilled raw window has seen every candidate and
+                # keeps the one-global-query fast path.
+                if (
+                    effective_limit
+                    and query_limit
+                    and (
+                        len(global_rows) >= query_limit
+                        or global_window_exhausted
+                    )
+                ):
                     # One GROUP BY over sessions.project_id (no join, no lineage
                     # recursion) says which projects still own rows this pass has
                     # not delivered, so the follow-up queries below are paid only